Fetch first support

Add support to ``FETCH {FIRST | NEXT} [ count ] {ROW | ROWS}
 {ONLY | WITH TIES}`` in the select for the supported backends,
currently PostgreSQL, Oracle and MSSQL.

Fixes: #5576
Change-Id: Ibb5871a457c0555f82b37e354e7787d15575f1f7

10 files changed, 415 insertions, 92 deletions

diff --git a/lib/sqlalchemy/dialects/mssql/base.py b/lib/sqlalchemy/dialects/mssql/base.py
index a2ed2b47d..c8f2b4ca3 100644
--- a/lib/sqlalchemy/dialects/mssql/base.py
+++ b/lib/sqlalchemy/dialects/mssql/base.py
@@ -1716,15 +1716,19 @@ class MSSQLCompiler(compiler.SQLCompiler):
 
         s = super(MSSQLCompiler, self).get_select_precolumns(select, **kw)
 
-        if select._simple_int_limit and (
-            select._offset_clause is None
-            or (select._simple_int_offset and select._offset == 0)
-        ):
+        if select._has_row_limiting_clause and self._use_top(select):
             # ODBC drivers and possibly others
             # don't support bind params in the SELECT clause on SQL Server.
             # so have to use literal here.
             kw["literal_execute"] = True
-            s += "TOP %s " % self.process(select._limit_clause, **kw)
+            s += "TOP %s " % self.process(
+                self._get_limit_or_fetch(select), **kw
+            )
+            if select._fetch_clause is not None:
+                if select._fetch_clause_options["percent"]:
+                    s += "PERCENT "
+                if select._fetch_clause_options["with_ties"]:
+                    s += "WITH TIES "
 
         return s
 
@@ -1734,27 +1738,65 @@ class MSSQLCompiler(compiler.SQLCompiler):
     def get_crud_hint_text(self, table, text):
         return text
 
-    def limit_clause(self, select, **kw):
+    def _get_limit_or_fetch(self, select):
+        if select._fetch_clause is None:
+            return select._limit_clause
+        else:
+            return select._fetch_clause
+
+    def _use_top(self, select):
+        return (
+            select._offset_clause is None
+            or (
+                select._simple_int_clause(select._offset_clause)
+                and select._offset == 0
+            )
+        ) and (
+            select._simple_int_clause(select._limit_clause)
+            or (
+                # limit can use TOP with is by itself. fetch only uses TOP
+                # when it needs to because of PERCENT and/or WITH TIES
+                select._simple_int_clause(select._fetch_clause)
+                and (
+                    select._fetch_clause_options["percent"]
+                    or select._fetch_clause_options["with_ties"]
+                )
+            )
+        )
+
+    def fetch_clause(self, cs, **kwargs):
+        return ""
+
+    def limit_clause(self, cs, **kwargs):
+        return ""
+
+    def _check_can_use_fetch_like(self, select):
+        # to use ROW_NUMBER(), an ORDER BY is required.
+        # OFFSET are FETCH are options of the ORDER BY clause
+        if not select._order_by_clause.clauses:
+            raise exc.CompileError(
+                "MSSQL requires an order_by when "
+                "using an OFFSET or a non-simple "
+                "LIMIT clause"
+            )
+
+        if select._fetch_clause_options is not None and (
+            select._fetch_clause_options["percent"]
+            or select._fetch_clause_options["with_ties"]
+        ):
+            raise exc.CompileError(
+                "MSSQL needs TOP to use PERCENT and/or WITH TIES. "
+                "Only simple fetch without offset can be used."
+            )
+
+    def _row_limit_clause(self, select, **kw):
         """MSSQL 2012 supports OFFSET/FETCH operators
         Use it instead subquery with row_number
 
         """
 
-        if self.dialect._supports_offset_fetch and (
-            (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
-            )
-        ):
-            # OFFSET are FETCH are options of the ORDER BY clause
-            if not select._order_by_clause.clauses:
-                raise exc.CompileError(
-                    "MSSQL requires an order_by when "
-                    "using an OFFSET or a non-simple "
-                    "LIMIT clause"
-                )
+        if self.dialect._supports_offset_fetch and not self._use_top(select):
+            self._check_can_use_fetch_like(select)
 
             text = ""
 
@@ -1764,9 +1806,11 @@ class MSSQLCompiler(compiler.SQLCompiler):
                 offset_str = "0"
             text += "\n OFFSET %s ROWS" % offset_str
 
-            if select._limit_clause is not None:
-                text += "\n FETCH NEXT %s ROWS ONLY " % self.process(
-                    select._limit_clause, **kw
+            limit = self._get_limit_or_fetch(select)
+
+            if limit is not None:
+                text += "\n FETCH FIRST %s ROWS ONLY" % self.process(
+                    limit, **kw
                 )
             return text
         else:
@@ -1787,35 +1831,19 @@ class MSSQLCompiler(compiler.SQLCompiler):
         select = select_stmt
 
         if (
-            not self.dialect._supports_offset_fetch
-            and (
-                (
-                    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
-                )
-            )
+            select._has_row_limiting_clause
+            and not self.dialect._supports_offset_fetch
+            and not self._use_top(select)
             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 or a non-simple "
-                    "LIMIT clause"
-                )
+            self._check_can_use_fetch_like(select)
 
             _order_by_clauses = [
                 sql_util.unwrap_label_reference(elem)
                 for elem in select._order_by_clause.clauses
             ]
 
-            limit_clause = select._limit_clause
+            limit_clause = self._get_limit_or_fetch(select)
             offset_clause = select._offset_clause
 
             select = select._generate()
diff --git a/lib/sqlalchemy/dialects/oracle/base.py b/lib/sqlalchemy/dialects/oracle/base.py
index fad1b0bbe..655614769 100644
--- a/lib/sqlalchemy/dialects/oracle/base.py
+++ b/lib/sqlalchemy/dialects/oracle/base.py
@@ -1037,9 +1037,13 @@ class OracleCompiler(compiler.SQLCompiler):
                     select = select.where(whereclause)
                     select._oracle_visit = True
 
-            limit_clause = select._limit_clause
-            offset_clause = select._offset_clause
-            if limit_clause is not None or offset_clause is not None:
+            # if fetch is used this is not needed
+            if (
+                select._has_row_limiting_clause
+                and select._fetch_clause is None
+            ):
+                limit_clause = select._limit_clause
+                offset_clause = select._offset_clause
                 # currently using form at:
                 # https://blogs.oracle.com/oraclemagazine/\
                 # on-rownum-and-limiting-results
@@ -1071,7 +1075,7 @@ class OracleCompiler(compiler.SQLCompiler):
                 if (
                     limit_clause is not None
                     and self.dialect.optimize_limits
-                    and select._simple_int_limit
+                    and select._simple_int_clause(limit_clause)
                 ):
                     param = sql.bindparam(
                         "_ora_frow",
@@ -1099,8 +1103,9 @@ class OracleCompiler(compiler.SQLCompiler):
 
                 # If needed, add the limiting clause
                 if limit_clause is not None:
-                    if select._simple_int_limit and (
-                        offset_clause is None or select._simple_int_offset
+                    if select._simple_int_clause(limit_clause) and (
+                        offset_clause is None
+                        or select._simple_int_clause(offset_clause)
                     ):
                         max_row = select._limit
 
@@ -1161,7 +1166,7 @@ class OracleCompiler(compiler.SQLCompiler):
                             adapter.traverse(elem) for elem in for_update.of
                         ]
 
-                    if select._simple_int_offset:
+                    if select._simple_int_clause(offset_clause):
                         offset_clause = sql.bindparam(
                             None,
                             select._offset,
diff --git a/lib/sqlalchemy/dialects/postgresql/asyncpg.py b/lib/sqlalchemy/dialects/postgresql/asyncpg.py
index 8b20de2b6..a4937d0d2 100644
--- a/lib/sqlalchemy/dialects/postgresql/asyncpg.py
+++ b/lib/sqlalchemy/dialects/postgresql/asyncpg.py
@@ -313,7 +313,6 @@ class AsyncAdapt_asyncpg_cursor:
 
         params = self._parameters()
         operation = re.sub(r"\?", lambda m: next(params), operation)
-
         try:
             prepared_stmt = await self._connection.prepare(operation)
 
diff --git a/lib/sqlalchemy/dialects/postgresql/base.py b/lib/sqlalchemy/dialects/postgresql/base.py
index 82f0126a0..44272b9d3 100644
--- a/lib/sqlalchemy/dialects/postgresql/base.py
+++ b/lib/sqlalchemy/dialects/postgresql/base.py
@@ -1953,7 +1953,7 @@ class PGCompiler(compiler.SQLCompiler):
             text += " \n LIMIT " + self.process(select._limit_clause, **kw)
         if select._offset_clause is not None:
             if select._limit_clause is None:
-                text += " \n LIMIT ALL"
+                text += "\n LIMIT ALL"
             text += " OFFSET " + self.process(select._offset_clause, **kw)
         return text
 
@@ -2142,6 +2142,25 @@ class PGCompiler(compiler.SQLCompiler):
             for t in extra_froms
         )
 
+    def fetch_clause(self, select, **kw):
+        # pg requires parens for non literal clauses. It's also required for
+        # bind parameters if a ::type casts is used by the driver (asyncpg),
+        # so it's easies to just always add it
+        text = ""
+        if select._offset_clause is not None:
+            text += "\n OFFSET (%s) ROWS" % self.process(
+                select._offset_clause, **kw
+            )
+        if select._fetch_clause is not None:
+            text += "\n FETCH FIRST (%s)%s ROWS %s" % (
+                self.process(select._fetch_clause, **kw),
+                " PERCENT" if select._fetch_clause_options["percent"] else "",
+                "WITH TIES"
+                if select._fetch_clause_options["with_ties"]
+                else "ONLY",
+            )
+        return text
+
 
 class PGDDLCompiler(compiler.DDLCompiler):
     def get_column_specification(self, column, **kwargs):
diff --git a/lib/sqlalchemy/orm/query.py b/lib/sqlalchemy/orm/query.py
index 0e7fd2fc3..5893a8b5b 100644
--- a/lib/sqlalchemy/orm/query.py
+++ b/lib/sqlalchemy/orm/query.py
@@ -300,6 +300,12 @@ class Query(
                 "are applied." % (meth, meth)
             )
 
+    @property
+    def _has_row_limiting_clause(self):
+        return (
+            self._limit_clause is not None or self._offset_clause is not None
+        )
+
     def _get_options(
         self,
         populate_existing=None,
diff --git a/lib/sqlalchemy/orm/strategies.py b/lib/sqlalchemy/orm/strategies.py
index 900691688..371f923ee 100644
--- a/lib/sqlalchemy/orm/strategies.py
+++ b/lib/sqlalchemy/orm/strategies.py
@@ -1253,7 +1253,7 @@ class SubqueryLoader(PostLoader):
                     break
 
         # don't need ORDER BY if no limit/offset
-        if q._limit_clause is None and q._offset_clause is None:
+        if not q._has_row_limiting_clause:
             q._order_by_clauses = ()
 
         if q._distinct is True and q._order_by_clauses:
diff --git a/lib/sqlalchemy/sql/compiler.py b/lib/sqlalchemy/sql/compiler.py
index 925441539..2fa9961eb 100644
--- a/lib/sqlalchemy/sql/compiler.py
+++ b/lib/sqlalchemy/sql/compiler.py
@@ -1745,11 +1745,8 @@ class SQLCompiler(Compiled):
         kwargs["include_table"] = False
         text += self.group_by_clause(cs, **dict(asfrom=asfrom, **kwargs))
         text += self.order_by_clause(cs, **kwargs)
-        text += (
-            (cs._limit_clause is not None or cs._offset_clause is not None)
-            and self.limit_clause(cs, **kwargs)
-            or ""
-        )
+        if cs._has_row_limiting_clause:
+            text += self._row_limit_clause(cs, **kwargs)
 
         if self.ctes and toplevel:
             text = self._render_cte_clause() + text
@@ -1757,6 +1754,12 @@ class SQLCompiler(Compiled):
         self.stack.pop(-1)
         return text
 
+    def _row_limit_clause(self, cs, **kwargs):
+        if cs._fetch_clause is not None:
+            return self.fetch_clause(cs, **kwargs)
+        else:
+            return self.limit_clause(cs, **kwargs)
+
     def _get_operator_dispatch(self, operator_, qualifier1, qualifier2):
         attrname = "visit_%s_%s%s" % (
             operator_.__name__,
@@ -3087,11 +3090,8 @@ class SQLCompiler(Compiled):
         if select._order_by_clauses:
             text += self.order_by_clause(select, **kwargs)
 
-        if (
-            select._limit_clause is not None
-            or select._offset_clause is not None
-        ):
-            text += self.limit_clause(select, **kwargs)
+        if select._has_row_limiting_clause:
+            text += self._row_limit_clause(select, **kwargs)
 
         if select._for_update_arg is not None:
             text += self.for_update_clause(select, **kwargs)
@@ -3183,6 +3183,22 @@ class SQLCompiler(Compiled):
             text += " OFFSET " + self.process(select._offset_clause, **kw)
         return text
 
+    def fetch_clause(self, select, **kw):
+        text = ""
+        if select._offset_clause is not None:
+            text += "\n OFFSET %s ROWS" % self.process(
+                select._offset_clause, **kw
+            )
+        if select._fetch_clause is not None:
+            text += "\n FETCH FIRST %s%s ROWS %s" % (
+                self.process(select._fetch_clause, **kw),
+                " PERCENT" if select._fetch_clause_options["percent"] else "",
+                "WITH TIES"
+                if select._fetch_clause_options["with_ties"]
+                else "ONLY",
+            )
+        return text
+
     def visit_table(
         self,
         table,
diff --git a/lib/sqlalchemy/sql/selectable.py b/lib/sqlalchemy/sql/selectable.py
index 5fc83815d..9a41d18e4 100644
--- a/lib/sqlalchemy/sql/selectable.py
+++ b/lib/sqlalchemy/sql/selectable.py
@@ -2804,6 +2804,8 @@ class GenerativeSelect(DeprecatedSelectBaseGenerations, SelectBase):
     _group_by_clauses = ()
     _limit_clause = None
     _offset_clause = None
+    _fetch_clause = None
+    _fetch_clause_options = None
     _for_update_arg = None
 
     def __init__(
@@ -2976,19 +2978,11 @@ class GenerativeSelect(DeprecatedSelectBaseGenerations, SelectBase):
         """
         return self._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
+    def _simple_int_clause(self, clause):
+        """True if the 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)
+        return isinstance(clause, _OffsetLimitParam)
 
     @property
     def _offset(self):
@@ -3002,6 +2996,14 @@ class GenerativeSelect(DeprecatedSelectBaseGenerations, SelectBase):
             self._offset_clause, "offset"
         )
 
+    @property
+    def _has_row_limiting_clause(self):
+        return (
+            self._limit_clause is not None
+            or self._offset_clause is not None
+            or self._fetch_clause is not None
+        )
+
     @_generative
     def limit(self, limit):
         """Return a new selectable with the given LIMIT criterion
@@ -3012,17 +3014,79 @@ class GenerativeSelect(DeprecatedSelectBaseGenerations, SelectBase):
         support ``LIMIT`` will attempt to provide similar
         functionality.
 
+        .. note::
+
+           The :meth:`_sql.GenerativeSelect.limit` method will replace
+           any clause applied with :meth:`_sql.GenerativeSelect.fetch`.
+
         .. versionchanged:: 1.0.0 - :meth:`_expression.Select.limit` can now
            accept arbitrary SQL expressions as well as integer values.
 
         :param limit: an integer LIMIT parameter, or a SQL expression
-         that provides an integer result.
+         that provides an integer result. Pass ``None`` to reset it.
+
+        .. seealso::
+
+           :meth:`_sql.GenerativeSelect.fetch`
+
+           :meth:`_sql.GenerativeSelect.offset`
 
         """
 
+        self._fetch_clause = self._fetch_clause_options = None
         self._limit_clause = self._offset_or_limit_clause(limit)
 
     @_generative
+    def fetch(self, count, with_ties=False, percent=False):
+        """Return a new selectable with the given FETCH FIRST criterion
+        applied.
+
+        This is a numeric value which usually renders as
+        ``FETCH {FIRST | NEXT} [ count ] {ROW | ROWS} {ONLY | WITH TIES}``
+        expression in the resulting select. This functionality is
+        is currently implemented for Oracle, PostgreSQL, MSSSQL.
+
+        Use :meth:`_sql.GenerativeSelect.offset` to specify the offset.
+
+        .. note::
+
+           The :meth:`_sql.GenerativeSelect.fetch` method will replace
+           any clause applied with :meth:`_sql.GenerativeSelect.limit`.
+
+        .. versionadded:: 1.4
+
+        :param count: an integer COUNT parameter, or a SQL expression
+         that provides an integer result. When ``percent=True`` this will
+         represent the percentage of rows to return, not the absolute value.
+         Pass ``None`` to reset it.
+
+        :param with_ties: When ``True``, the WITH TIES option is used
+         to return any additional rows that tie for the last place in the
+         result set according to the ``ORDER BY`` clause. The
+         ``ORDER BY`` may be mandatory in this case. Defaults to ``False``
+
+        :param percent: When ``True``, ``count`` represents the percentage
+         of the total number of selected rows to return. Defaults to ``False``
+
+        .. seealso::
+
+           :meth:`_sql.GenerativeSelect.limit`
+
+           :meth:`_sql.GenerativeSelect.offset`
+
+        """
+
+        self._limit_clause = None
+        if count is None:
+            self._fetch_clause = self._fetch_clause_options = None
+        else:
+            self._fetch_clause = self._offset_or_limit_clause(count)
+            self._fetch_clause_options = {
+                "with_ties": with_ties,
+                "percent": percent,
+            }
+
+    @_generative
     def offset(self, offset):
         """Return a new selectable with the given OFFSET criterion
         applied.
@@ -3038,7 +3102,13 @@ class GenerativeSelect(DeprecatedSelectBaseGenerations, SelectBase):
            accept arbitrary SQL expressions as well as integer values.
 
         :param offset: an integer OFFSET parameter, or a SQL expression
-         that provides an integer result.
+         that provides an integer result. Pass ``None`` to reset it.
+
+        .. seealso::
+
+           :meth:`_sql.GenerativeSelect.limit`
+
+           :meth:`_sql.GenerativeSelect.fetch`
 
         """
 
@@ -3068,6 +3138,11 @@ class GenerativeSelect(DeprecatedSelectBaseGenerations, SelectBase):
            LIMIT ? OFFSET ?
            (2, 1)
 
+        .. note::
+
+           The :meth:`_sql.GenerativeSelect.slice` method will replace
+           any clause applied with :meth:`_sql.GenerativeSelect.fetch`.
+
         .. versionadded:: 1.4  Added the :meth:`_sql.GenerativeSelect.slice`
            method generalized from the ORM.
 
@@ -3077,8 +3152,11 @@ class GenerativeSelect(DeprecatedSelectBaseGenerations, SelectBase):
 
            :meth:`_sql.GenerativeSelect.offset`
 
+           :meth:`_sql.GenerativeSelect.fetch`
+
         """
         sql_util = util.preloaded.sql_util
+        self._fetch_clause = self._fetch_clause_options = None
         self._limit_clause, self._offset_clause = sql_util._make_slice(
             self._limit_clause, self._offset_clause, start, stop
         )
@@ -3177,6 +3255,8 @@ class CompoundSelect(HasCompileState, GenerativeSelect):
         ("selects", InternalTraversal.dp_clauseelement_list),
         ("_limit_clause", InternalTraversal.dp_clauseelement),
         ("_offset_clause", InternalTraversal.dp_clauseelement),
+        ("_fetch_clause", InternalTraversal.dp_clauseelement),
+        ("_fetch_clause_options", InternalTraversal.dp_plain_dict),
         ("_order_by_clauses", InternalTraversal.dp_clauseelement_list),
         ("_group_by_clauses", InternalTraversal.dp_clauseelement_list),
         ("_for_update_arg", InternalTraversal.dp_clauseelement),
@@ -3933,6 +4013,8 @@ class Select(
             ("_correlate_except", InternalTraversal.dp_clauseelement_tuple),
             ("_limit_clause", InternalTraversal.dp_clauseelement),
             ("_offset_clause", InternalTraversal.dp_clauseelement),
+            ("_fetch_clause", InternalTraversal.dp_clauseelement),
+            ("_fetch_clause_options", InternalTraversal.dp_plain_dict),
             ("_for_update_arg", InternalTraversal.dp_clauseelement),
             ("_distinct", InternalTraversal.dp_boolean),
             ("_distinct_on", InternalTraversal.dp_clauseelement_tuple),
@@ -5202,10 +5284,8 @@ class Select(
         subquery._columns._populate_separate_keys(prox)
 
     def _needs_parens_for_grouping(self):
-        return (
-            self._limit_clause is not None
-            or self._offset_clause is not None
-            or bool(self._order_by_clause.clauses)
+        return self._has_row_limiting_clause or bool(
+            self._order_by_clause.clauses
         )
 
     def self_group(self, against=None):
diff --git a/lib/sqlalchemy/testing/requirements.py b/lib/sqlalchemy/testing/requirements.py
index b7f0d0f59..45a2fdf31 100644
--- a/lib/sqlalchemy/testing/requirements.py
+++ b/lib/sqlalchemy/testing/requirements.py
@@ -1292,19 +1292,37 @@ class SuiteRequirements(Requirements):
 
     @property
     def regexp_match(self):
-        """backend supports the regexp_match operator.
+        """backend supports the regexp_match operator."""
+        return exclusions.closed()
 
-        .. versionadded:: 1.4
+    @property
+    def regexp_replace(self):
+        """backend supports the regexp_replace operator."""
+        return exclusions.closed()
 
-        """
+    @property
+    def fetch_first(self):
+        """backend supports the fetch first clause."""
         return exclusions.closed()
 
     @property
-    def regexp_replace(self):
-        """backend supports the regexp_replace operator.
+    def fetch_percent(self):
+        """backend supports the fetch first clause with percent."""
+        return exclusions.closed()
 
-        .. versionadded:: 1.4
+    @property
+    def fetch_ties(self):
+        """backend supports the fetch first clause with ties."""
+        return exclusions.closed()
 
+    @property
+    def fetch_no_order_by(self):
+        """backend supports the fetch first without order by"""
+        return exclusions.closed()
 
+    @property
+    def fetch_offset_with_options(self):
+        """backend supports the offset when using fetch first with percent
+        or ties. basically this is "not mssql"
         """
         return exclusions.closed()
diff --git a/lib/sqlalchemy/testing/suite/test_select.py b/lib/sqlalchemy/testing/suite/test_select.py
index a2924bfde..ee9db9111 100644
--- a/lib/sqlalchemy/testing/suite/test_select.py
+++ b/lib/sqlalchemy/testing/suite/test_select.py
@@ -151,7 +151,7 @@ class OrderByLabelTest(fixtures.TablesTest):
         self._assert_result(stmt, [(1, 3), (1, 5), (1, 7)])
 
 
-class LimitOffsetTest(fixtures.TablesTest):
+class FetchLimitOffsetTest(fixtures.TablesTest):
     __backend__ = True
 
     @classmethod
@@ -173,6 +173,7 @@ class LimitOffsetTest(fixtures.TablesTest):
                 {"id": 2, "x": 2, "y": 3},
                 {"id": 3, "x": 3, "y": 4},
                 {"id": 4, "x": 4, "y": 5},
+                {"id": 5, "x": 4, "y": 6},
             ],
         )
 
@@ -191,12 +192,20 @@ class LimitOffsetTest(fixtures.TablesTest):
             [(1, 1, 2), (2, 2, 3)],
         )
 
+    @testing.requires.fetch_first
+    def test_simple_fetch(self):
+        table = self.tables.some_table
+        self._assert_result(
+            select(table).order_by(table.c.id).fetch(2),
+            [(1, 1, 2), (2, 2, 3)],
+        )
+
     @testing.requires.offset
     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)],
+            [(3, 3, 4), (4, 4, 5), (5, 4, 6)],
         )
 
     @testing.requires.offset
@@ -207,6 +216,23 @@ class LimitOffsetTest(fixtures.TablesTest):
             [(2, 2, 3), (3, 3, 4)],
         )
 
+    @testing.requires.fetch_first
+    def test_simple_fetch_offset(self):
+        table = self.tables.some_table
+        self._assert_result(
+            select(table).order_by(table.c.id).fetch(2).offset(1),
+            [(2, 2, 3), (3, 3, 4)],
+        )
+
+    @testing.requires.fetch_no_order_by
+    def test_fetch_offset_no_order(self):
+        table = self.tables.some_table
+        with config.db.connect() as conn:
+            eq_(
+                set(conn.execute(select(table).fetch(10))),
+                set([(1, 1, 2), (2, 2, 3), (3, 3, 4), (4, 4, 5), (5, 4, 6)]),
+            )
+
     @testing.requires.offset
     def test_limit_offset_nobinds(self):
         """test that 'literal binds' mode works - no bound params."""
@@ -220,6 +246,19 @@ class LimitOffsetTest(fixtures.TablesTest):
 
         self._assert_result_str(sql, [(2, 2, 3), (3, 3, 4)])
 
+    @testing.requires.fetch_first
+    def test_fetch_offset_nobinds(self):
+        """test that 'literal binds' mode works - no bound params."""
+
+        table = self.tables.some_table
+        stmt = select(table).order_by(table.c.id).fetch(2).offset(1)
+        sql = stmt.compile(
+            dialect=config.db.dialect, compile_kwargs={"literal_binds": True}
+        )
+        sql = str(sql)
+
+        self._assert_result_str(sql, [(2, 2, 3), (3, 3, 4)])
+
     @testing.requires.bound_limit_offset
     def test_bound_limit(self):
         table = self.tables.some_table
@@ -234,7 +273,7 @@ class LimitOffsetTest(fixtures.TablesTest):
         table = self.tables.some_table
         self._assert_result(
             select(table).order_by(table.c.id).offset(bindparam("o")),
-            [(3, 3, 4), (4, 4, 5)],
+            [(3, 3, 4), (4, 4, 5), (5, 4, 6)],
             params={"o": 2},
         )
 
@@ -250,6 +289,18 @@ class LimitOffsetTest(fixtures.TablesTest):
             params={"l": 2, "o": 1},
         )
 
+    @testing.requires.fetch_first
+    def test_bound_fetch_offset(self):
+        table = self.tables.some_table
+        self._assert_result(
+            select(table)
+            .order_by(table.c.id)
+            .fetch(bindparam("f"))
+            .offset(bindparam("o")),
+            [(2, 2, 3), (3, 3, 4)],
+            params={"f": 2, "o": 1},
+        )
+
     @testing.requires.sql_expression_limit_offset
     def test_expr_offset(self):
         table = self.tables.some_table
@@ -257,7 +308,7 @@ class LimitOffsetTest(fixtures.TablesTest):
             select(table)
             .order_by(table.c.id)
             .offset(literal_column("1") + literal_column("2")),
-            [(4, 4, 5)],
+            [(4, 4, 5), (5, 4, 6)],
         )
 
     @testing.requires.sql_expression_limit_offset
@@ -281,6 +332,17 @@ class LimitOffsetTest(fixtures.TablesTest):
             [(3, 3, 4), (4, 4, 5)],
         )
 
+    @testing.requires.fetch_first
+    def test_expr_fetch_offset(self):
+        table = self.tables.some_table
+        self._assert_result(
+            select(table)
+            .order_by(table.c.id)
+            .fetch(literal_column("1") + literal_column("1"))
+            .offset(literal_column("1") + literal_column("1")),
+            [(3, 3, 4), (4, 4, 5)],
+        )
+
     @testing.requires.sql_expression_limit_offset
     def test_simple_limit_expr_offset(self):
         table = self.tables.some_table
@@ -303,6 +365,96 @@ class LimitOffsetTest(fixtures.TablesTest):
             [(3, 3, 4), (4, 4, 5)],
         )
 
+    @testing.requires.fetch_ties
+    def test_simple_fetch_ties(self):
+        table = self.tables.some_table
+        with config.db.connect() as conn:
+            eq_(
+                set(
+                    conn.execute(
+                        select(table)
+                        .order_by(table.c.x.desc())
+                        .fetch(1, with_ties=True)
+                    )
+                ),
+                set([(4, 4, 5), (5, 4, 6)]),
+            )
+
+    @testing.requires.fetch_ties
+    @testing.requires.fetch_offset_with_options
+    def test_fetch_offset_ties(self):
+        table = self.tables.some_table
+        with config.db.connect() as conn:
+            fa = conn.execute(
+                select(table)
+                .order_by(table.c.x)
+                .fetch(2, with_ties=True)
+                .offset(2)
+            ).fetchall()
+            eq_(fa[0], (3, 3, 4))
+            eq_(set(fa), set([(3, 3, 4), (4, 4, 5), (5, 4, 6)]))
+
+    @testing.requires.fetch_ties
+    @testing.requires.fetch_offset_with_options
+    def test_fetch_offset_ties_exact_number(self):
+        table = self.tables.some_table
+        self._assert_result(
+            select(table)
+            .order_by(table.c.x)
+            .fetch(2, with_ties=True)
+            .offset(1),
+            [(2, 2, 3), (3, 3, 4)],
+        )
+
+    @testing.requires.fetch_percent
+    def test_simple_fetch_percent(self):
+        table = self.tables.some_table
+        self._assert_result(
+            select(table).order_by(table.c.id).fetch(20, percent=True),
+            [(1, 1, 2)],
+        )
+
+    @testing.requires.fetch_percent
+    @testing.requires.fetch_offset_with_options
+    def test_fetch_offset_percent(self):
+        table = self.tables.some_table
+        self._assert_result(
+            select(table)
+            .order_by(table.c.id)
+            .fetch(40, percent=True)
+            .offset(1),
+            [(2, 2, 3), (3, 3, 4)],
+        )
+
+    @testing.requires.fetch_ties
+    @testing.requires.fetch_percent
+    def test_simple_fetch_percent_ties(self):
+        table = self.tables.some_table
+        with config.db.connect() as conn:
+            fa = conn.execute(
+                select(table)
+                .order_by(table.c.x.desc())
+                .fetch(20, percent=True, with_ties=True)
+            ).fetchall()
+
+        eq_(len(fa), 2)
+        eq_(set(fa), set([(4, 4, 5), (5, 4, 6)]))
+
+    @testing.requires.fetch_ties
+    @testing.requires.fetch_percent
+    @testing.requires.fetch_offset_with_options
+    def test_fetch_offset_percent_ties(self):
+        table = self.tables.some_table
+        with config.db.connect() as conn:
+            fa = conn.execute(
+                select(table)
+                .order_by(table.c.x)
+                .fetch(40, percent=True, with_ties=True)
+                .offset(2)
+            ).fetchall()
+            eq_(fa[0], (3, 3, 4))
+            eq_(set(fa), set([(3, 3, 4), (4, 4, 5), (5, 4, 6)]))
+
 
 class JoinTest(fixtures.TablesTest):
     __backend__ = True