Fixed orm not applying fetch

Fixed an issue where :meth:`_sql.GenerativeSelect.fetch` would be
ignored when executing a statement using the ORM.

Fixes: #8091
Change-Id: I6790c7272a71278e90de2529c8bc8ae89e54e288

4 files changed, 44 insertions, 1 deletions

diff --git a/doc/build/changelog/unreleased_14/8091.rst b/doc/build/changelog/unreleased_14/8091.rst
new file mode 100644
index 000000000..014f66a56
--- /dev/null
+++ b/doc/build/changelog/unreleased_14/8091.rst
@@ -0,0 +1,6 @@
+.. change::
+    :tags: bug, orm, sql
+    :tickets: 8091
+
+    Fixed an issue where :meth:`_sql.GenerativeSelect.fetch` would not
+    be applied when executing a statement using the ORM.
diff --git a/lib/sqlalchemy/dialects/sqlite/aiosqlite.py b/lib/sqlalchemy/dialects/sqlite/aiosqlite.py
index a5a1ec065..93c1a499d 100644
--- a/lib/sqlalchemy/dialects/sqlite/aiosqlite.py
+++ b/lib/sqlalchemy/dialects/sqlite/aiosqlite.py
@@ -212,7 +212,6 @@ class AsyncAdapt_aiosqlite_connection(AdaptedConnection):
             self._handle_exception(error)
 
     def close(self):
-        # print(">close", self)
         try:
             self.await_(self._connection.close())
         except Exception as error:
diff --git a/lib/sqlalchemy/orm/context.py b/lib/sqlalchemy/orm/context.py
index 58556bb58..8676f828e 100644
--- a/lib/sqlalchemy/orm/context.py
+++ b/lib/sqlalchemy/orm/context.py
@@ -1338,6 +1338,8 @@ class ORMSelectCompileState(ORMCompileState, SelectState):
         correlate_except,
         limit_clause,
         offset_clause,
+        fetch_clause,
+        fetch_clause_options,
         distinct,
         distinct_on,
         prefixes,
@@ -1369,6 +1371,8 @@ class ORMSelectCompileState(ORMCompileState, SelectState):
 
         statement._limit_clause = limit_clause
         statement._offset_clause = offset_clause
+        statement._fetch_clause = fetch_clause
+        statement._fetch_clause_options = fetch_clause_options
 
         if prefixes:
             statement._prefixes = prefixes
@@ -2037,6 +2041,10 @@ class ORMSelectCompileState(ORMCompileState, SelectState):
             "prefixes": self.select_statement._prefixes,
             "suffixes": self.select_statement._suffixes,
             "group_by": self.group_by or None,
+            "fetch_clause": self.select_statement._fetch_clause,
+            "fetch_clause_options": (
+                self.select_statement._fetch_clause_options
+            ),
         }
 
     @property
diff --git a/test/orm/test_core_compilation.py b/test/orm/test_core_compilation.py
index 631509ca0..19b1c34b1 100644
--- a/test/orm/test_core_compilation.py
+++ b/test/orm/test_core_compilation.py
@@ -304,6 +304,36 @@ class SelectableTest(QueryTest, AssertsCompiledSQL):
         stmt = testing.resolve_lambda(stmt, **lambda_args)
         self.assert_compile(stmt, expected)
 
+    def test_limit_offset_select(self):
+        User = self.classes.User
+
+        stmt = select(User.id).limit(5).offset(6)
+        self.assert_compile(
+            stmt,
+            "SELECT users.id FROM users LIMIT :param_1 OFFSET :param_2",
+            checkparams={"param_1": 5, "param_2": 6},
+        )
+
+    @testing.combinations(
+        (None, "ROWS ONLY"),
+        ({"percent": True}, "PERCENT ROWS ONLY"),
+        ({"percent": True, "with_ties": True}, "PERCENT ROWS WITH TIES"),
+    )
+    def test_fetch_offset_select(self, options, fetch_clause):
+        User = self.classes.User
+
+        if options is None:
+            stmt = select(User.id).fetch(5).offset(6)
+        else:
+            stmt = select(User.id).fetch(5, **options).offset(6)
+
+        self.assert_compile(
+            stmt,
+            "SELECT users.id FROM users OFFSET :param_1 "
+            "ROWS FETCH FIRST :param_2 %s" % (fetch_clause,),
+            checkparams={"param_1": 6, "param_2": 5},
+        )
+
 
 class ColumnsClauseFromsTest(QueryTest, AssertsCompiledSQL):
     __dialect__ = "default"