Render LIMIT/OFFSET conditions after compile on select dialects

Added new "post compile parameters" feature.  This feature allows a
:func:`.bindparam` construct to have its value rendered into the SQL string
before being passed to the DBAPI driver, but after the compilation step,
using the "literal render" feature of the compiler.  The immediate
rationale for this feature is to support LIMIT/OFFSET schemes that don't
work or perform well as bound parameters handled by the database driver,
while still allowing for SQLAlchemy SQL constructs to be cacheable in their
compiled form.     The immediate targets for the new feature are the "TOP
N" clause used by SQL Server (and Sybase) which does not support a bound
parameter, as well as the "ROWNUM" and optional "FIRST_ROWS()" schemes used
by the Oracle dialect, the former of which has been known to perform better
without bound parameters and the latter of which does not support a bound
parameter.   The feature builds upon the mechanisms first developed to
support "expanding" parameters for IN expressions.   As part of this
feature, the Oracle ``use_binds_for_limits`` feature is turned on
unconditionally and this flag is now deprecated.

- adds limited support for "unique" bound parameters within
a text() construct.

- adds an additional int() check within the literal render
function of the Integer datatype and tests that non-int values
raise ValueError.

Fixes: #4808
Change-Id: Iace97d544d1a7351ee07db970c6bc06a19c712c6

3 files changed, 212 insertions, 3 deletions

diff --git a/lib/sqlalchemy/testing/assertsql.py b/lib/sqlalchemy/testing/assertsql.py
index 00496d549..55f4dc2ab 100644
--- a/lib/sqlalchemy/testing/assertsql.py
+++ b/lib/sqlalchemy/testing/assertsql.py
@@ -38,11 +38,10 @@ class SQLMatchRule(AssertRule):
 
 
 class CursorSQL(SQLMatchRule):
-    consume_statement = False
-
-    def __init__(self, statement, params=None):
+    def __init__(self, statement, params=None, consume_statement=True):
         self.statement = statement
         self.params = params
+        self.consume_statement = consume_statement
 
     def process_statement(self, execute_observed):
         stmt = execute_observed.statements[0]
diff --git a/lib/sqlalchemy/testing/requirements.py b/lib/sqlalchemy/testing/requirements.py
index 62e600eaa..27def1004 100644
--- a/lib/sqlalchemy/testing/requirements.py
+++ b/lib/sqlalchemy/testing/requirements.py
@@ -45,6 +45,26 @@ class SuiteRequirements(Requirements):
         return exclusions.open()
 
     @property
+    def standard_cursor_sql(self):
+        """Target database passes SQL-92 style statements to cursor.execute()
+        when a statement like select() or insert() is run.
+
+        A very small portion of dialect-level tests will ensure that certain
+        conditions are present in SQL strings, and these tests use very basic
+        SQL that will work on any SQL-like platform in order to assert results.
+
+        It's normally a given for any pep-249 DBAPI that a statement like
+        "SELECT id, name FROM table WHERE some_table.id=5" will work.
+        However, there are dialects that don't actually produce SQL Strings
+        and instead may work with symbolic objects instead, or dialects that
+        aren't working with SQL, so for those this requirement can be marked
+        as excluded.
+
+        """
+
+        return exclusions.open()
+
+    @property
     def on_update_cascade(self):
         """"target database must support ON UPDATE..CASCADE behavior in
         foreign keys."""
diff --git a/lib/sqlalchemy/testing/suite/test_select.py b/lib/sqlalchemy/testing/suite/test_select.py
index dabb30f9b..02cdcf4f5 100644
--- a/lib/sqlalchemy/testing/suite/test_select.py
+++ b/lib/sqlalchemy/testing/suite/test_select.py
@@ -1,7 +1,10 @@
+from .. import AssertsCompiledSQL
+from .. import AssertsExecutionResults
 from .. import config
 from .. import fixtures
 from ..assertions import eq_
 from ..assertions import in_
+from ..assertsql import CursorSQL
 from ..schema import Column
 from ..schema import Table
 from ... import bindparam
@@ -14,6 +17,7 @@ from ... import null
 from ... import select
 from ... import String
 from ... import testing
+from ... import text
 from ... import true
 from ... import tuple_
 from ... import union
@@ -233,6 +237,59 @@ class LimitOffsetTest(fixtures.TablesTest):
             params={"l": 2, "o": 1},
         )
 
+    @testing.requires.sql_expression_limit_offset
+    def test_expr_offset(self):
+        table = self.tables.some_table
+        self._assert_result(
+            select([table])
+            .order_by(table.c.id)
+            .offset(literal_column("1") + literal_column("2")),
+            [(4, 4, 5)],
+        )
+
+    @testing.requires.sql_expression_limit_offset
+    def test_expr_limit(self):
+        table = self.tables.some_table
+        self._assert_result(
+            select([table])
+            .order_by(table.c.id)
+            .limit(literal_column("1") + literal_column("2")),
+            [(1, 1, 2), (2, 2, 3), (3, 3, 4)],
+        )
+
+    @testing.requires.sql_expression_limit_offset
+    def test_expr_limit_offset(self):
+        table = self.tables.some_table
+        self._assert_result(
+            select([table])
+            .order_by(table.c.id)
+            .limit(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
+        self._assert_result(
+            select([table])
+            .order_by(table.c.id)
+            .limit(2)
+            .offset(literal_column("1") + literal_column("1")),
+            [(3, 3, 4), (4, 4, 5)],
+        )
+
+    @testing.requires.sql_expression_limit_offset
+    def test_expr_limit_simple_offset(self):
+        table = self.tables.some_table
+        self._assert_result(
+            select([table])
+            .order_by(table.c.id)
+            .limit(literal_column("1") + literal_column("1"))
+            .offset(2),
+            [(3, 3, 4), (4, 4, 5)],
+        )
+
 
 class CompoundSelectTest(fixtures.TablesTest):
     __backend__ = True
@@ -372,6 +429,127 @@ class CompoundSelectTest(fixtures.TablesTest):
         )
 
 
+class PostCompileParamsTest(
+    AssertsExecutionResults, AssertsCompiledSQL, fixtures.TablesTest
+):
+    __backend__ = True
+
+    __requires__ = ("standard_cursor_sql",)
+
+    @classmethod
+    def define_tables(cls, metadata):
+        Table(
+            "some_table",
+            metadata,
+            Column("id", Integer, primary_key=True),
+            Column("x", Integer),
+            Column("y", Integer),
+            Column("z", String(50)),
+        )
+
+    @classmethod
+    def insert_data(cls):
+        config.db.execute(
+            cls.tables.some_table.insert(),
+            [
+                {"id": 1, "x": 1, "y": 2, "z": "z1"},
+                {"id": 2, "x": 2, "y": 3, "z": "z2"},
+                {"id": 3, "x": 3, "y": 4, "z": "z3"},
+                {"id": 4, "x": 4, "y": 5, "z": "z4"},
+            ],
+        )
+
+    def test_compile(self):
+        table = self.tables.some_table
+
+        stmt = select([table.c.id]).where(
+            table.c.x == bindparam("q", literal_execute=True)
+        )
+
+        self.assert_compile(
+            stmt,
+            "SELECT some_table.id FROM some_table "
+            "WHERE some_table.x = [POSTCOMPILE_q]",
+            {},
+        )
+
+    def test_compile_literal_binds(self):
+        table = self.tables.some_table
+
+        stmt = select([table.c.id]).where(
+            table.c.x == bindparam("q", 10, literal_execute=True)
+        )
+
+        self.assert_compile(
+            stmt,
+            "SELECT some_table.id FROM some_table WHERE some_table.x = 10",
+            {},
+            literal_binds=True,
+        )
+
+    def test_execute(self):
+        table = self.tables.some_table
+
+        stmt = select([table.c.id]).where(
+            table.c.x == bindparam("q", literal_execute=True)
+        )
+
+        with self.sql_execution_asserter() as asserter:
+            with config.db.connect() as conn:
+                conn.execute(stmt, q=10)
+
+        asserter.assert_(
+            CursorSQL(
+                "SELECT some_table.id \nFROM some_table "
+                "\nWHERE some_table.x = 10",
+                () if config.db.dialect.positional else {},
+            )
+        )
+
+    def test_execute_expanding_plus_literal_execute(self):
+        table = self.tables.some_table
+
+        stmt = select([table.c.id]).where(
+            table.c.x.in_(bindparam("q", expanding=True, literal_execute=True))
+        )
+
+        with self.sql_execution_asserter() as asserter:
+            with config.db.connect() as conn:
+                conn.execute(stmt, q=[5, 6, 7])
+
+        asserter.assert_(
+            CursorSQL(
+                "SELECT some_table.id \nFROM some_table "
+                "\nWHERE some_table.x IN (5, 6, 7)",
+                () if config.db.dialect.positional else {},
+            )
+        )
+
+    @testing.requires.tuple_in
+    def test_execute_tuple_expanding_plus_literal_execute(self):
+        table = self.tables.some_table
+
+        stmt = select([table.c.id]).where(
+            tuple_(table.c.x, table.c.y).in_(
+                bindparam("q", expanding=True, literal_execute=True)
+            )
+        )
+
+        with self.sql_execution_asserter() as asserter:
+            with config.db.connect() as conn:
+                conn.execute(stmt, q=[(5, 10), (12, 18)])
+
+        asserter.assert_(
+            CursorSQL(
+                "SELECT some_table.id \nFROM some_table "
+                "\nWHERE (some_table.x, some_table.y) "
+                "IN (%s(5, 10), (12, 18))"
+                % ("VALUES " if config.db.dialect.tuple_in_values else ""),
+                () if config.db.dialect.positional else {},
+            )
+        )
+
+
 class ExpandingBoundInTest(fixtures.TablesTest):
     __backend__ = True
 
@@ -496,6 +674,18 @@ class ExpandingBoundInTest(fixtures.TablesTest):
             params={"q": [(2, "z2"), (3, "z3"), (4, "z4")]},
         )
 
+    @testing.requires.tuple_in
+    def test_bound_in_heterogeneous_two_tuple_text(self):
+        stmt = text(
+            "select id FROM some_table WHERE (x, z) IN :q ORDER BY id"
+        ).bindparams(bindparam("q", expanding=True))
+
+        self._assert_result(
+            stmt,
+            [(2,), (3,), (4,)],
+            params={"q": [(2, "z2"), (3, "z3"), (4, "z4")]},
+        )
+
     def test_empty_set_against_integer(self):
         table = self.tables.some_table