uniquify when popping literal_execute_params from param dict

Fixed further issues in the same area as that of :ticket:`6173` released in
1.4.5, where a "postcompile" parameter, again most typically those used for
LIMIT/OFFSET rendering in Oracle and SQL Server, would fail to be processed
correctly if the same parameter rendered in multiple places in the
statement.

Fixes: #6202
Change-Id: I95c355aa52a7546fe579ad67f9a8402a213cb79d

diff --git a/doc/build/changelog/unreleased_14/6202.rst b/doc/build/changelog/unreleased_14/6202.rst
new file mode 100644 (file)
index 0000000..9527e06
--- /dev/null
+++ b/doc/build/changelog/unreleased_14/6202.rst
@@ -0,0 +1,11 @@
+.. change::
+    :tags: bug, sql, oracle, mssql
+    :tickets: 6202
+
+    Fixed further issues in the same area as that of :ticket:`6173` released in
+    1.4.5, where a "postcompile" parameter, again most typically those used for
+    LIMIT/OFFSET rendering in Oracle and SQL Server, would fail to be processed
+    correctly if the same parameter rendered in multiple places in the
+    statement.
+
+

diff --git a/lib/sqlalchemy/sql/compiler.py b/lib/sqlalchemy/sql/compiler.py
index 32530629b467397d1377e42d13cf3d07c5fe8483..61dfbeb07dc9ed0fe83701eaf54c2d9c29e7bc4e 100644 (file)
--- a/lib/sqlalchemy/sql/compiler.py
+++ b/lib/sqlalchemy/sql/compiler.py
@@ -1137,7 +1137,9 @@ class SQLCompiler(Compiled):
         ):
             parameter = self.binds[name]
             if parameter in self.literal_execute_params:
-                value = parameters.pop(name)
+                if name not in replacement_expressions:
+                    value = parameters.pop(name)
+
                 replacement_expressions[name] = self.render_literal_bindparam(
                     parameter, render_literal_value=value
                 )

diff --git a/lib/sqlalchemy/testing/suite/test_select.py b/lib/sqlalchemy/testing/suite/test_select.py
index c519154f6f195ad77d3b9207e0f69d8451a2e592..7318a4f33a994498b19b31afb06254246a5210e2 100644 (file)
--- a/lib/sqlalchemy/testing/suite/test_select.py
+++ b/lib/sqlalchemy/testing/suite/test_select.py
@@ -223,6 +223,20 @@ class FetchLimitOffsetTest(fixtures.TablesTest):
             [(1, 1, 2), (2, 2, 3), (3, 3, 4)],
         )
 
+    def test_limit_render_multiple_times(self, connection):
+        table = self.tables.some_table
+        stmt = select(table.c.id).limit(1).scalar_subquery()
+
+        u = union(select(stmt), select(stmt)).subquery().select()
+
+        self._assert_result(
+            connection,
+            u,
+            [
+                (1,),
+            ],
+        )
+
     @testing.requires.fetch_first
     def test_simple_fetch(self, connection):
         table = self.tables.some_table

diff --git a/test/sql/test_compiler.py b/test/sql/test_compiler.py
index b2d44343849896c426234d79756a9fd6ce6c2c9f..2115c32a7e2e09d83ce27d45783295b02551d6f0 100644 (file)
--- a/test/sql/test_compiler.py
+++ b/test/sql/test_compiler.py
@@ -3964,6 +3964,91 @@ class BindParameterTest(AssertsCompiledSQL, fixtures.TestBase):
             "(:param_1_2_1, :param_1_2_2)",
         )
 
+    def test_construct_params_repeated_postcompile_params_one(self):
+        """test for :ticket:`6202` one - name repeated in positiontup
+        (e.g. SQL Server using TOP)
+
+        """
+
+        t = table("t", column("x"))
+        stmt = (
+            select(1)
+            .where(t.c.x == bindparam(None, value="10", literal_execute=True))
+            .scalar_subquery()
+        )
+
+        u = union(select(stmt), select(stmt)).subquery().select()
+
+        compiled = u.compile(
+            dialect=default.DefaultDialect(paramstyle="format"),
+            compile_kwargs={"render_postcompile": True},
+        )
+        eq_ignore_whitespace(
+            compiled.string,
+            "SELECT anon_2.anon_1 FROM (SELECT (SELECT 1 FROM t "
+            "WHERE t.x = '10') AS anon_1 UNION SELECT "
+            "(SELECT 1 FROM t WHERE t.x = '10') AS anon_1) AS anon_2",
+        )
+        eq_(compiled.construct_params(), {"param_1": "10"})
+
+    def test_construct_params_repeated_postcompile_params_two(self):
+        """test for :ticket:`6202` two - same param name used twice
+        (e.g. Oracle LIMIT)
+
+        """
+        t = table("t", column("x"))
+
+        bp = bindparam(None, value="10")
+        stmt = (
+            select(1)
+            .where(t.c.x == bp.render_literal_execute())
+            .scalar_subquery()
+        )
+        stmt2 = (
+            select(1)
+            .where(t.c.x == bp.render_literal_execute())
+            .scalar_subquery()
+        )
+
+        u = union(select(stmt), select(stmt2)).subquery().select()
+
+        compiled = u.compile(
+            dialect=default.DefaultDialect(paramstyle="named"),
+            compile_kwargs={"render_postcompile": True},
+        )
+        eq_ignore_whitespace(
+            compiled.string,
+            "SELECT anon_2.anon_1 FROM (SELECT (SELECT 1 "
+            "FROM t WHERE t.x = '10') AS anon_1 UNION SELECT "
+            "(SELECT 1 FROM t WHERE t.x = '10') AS anon_3) AS anon_2",
+        )
+        eq_(compiled.construct_params(), {"param_1": "10"})
+
+    def test_construct_params_positional_plain_repeated(self):
+        t = table("t", column("x"))
+        stmt = (
+            select(1)
+            .where(t.c.x == bindparam(None, value="10"))
+            .where(t.c.x == bindparam(None, value="12", literal_execute=True))
+            .scalar_subquery()
+        )
+
+        u = union(select(stmt), select(stmt)).subquery().select()
+
+        compiled = u.compile(
+            dialect=default.DefaultDialect(paramstyle="format"),
+            compile_kwargs={"render_postcompile": True},
+        )
+        eq_ignore_whitespace(
+            compiled.string,
+            "SELECT anon_2.anon_1 FROM (SELECT (SELECT 1 FROM t "
+            "WHERE t.x = %s AND t.x = '12') AS anon_1 "
+            "UNION SELECT (SELECT 1 FROM t WHERE t.x = %s AND t.x = '12') "
+            "AS anon_1) AS anon_2",
+        )
+        eq_(compiled.construct_params(), {"param_1": "10", "param_2": "12"})
+        eq_(compiled.positiontup, ["param_1", "param_1"])
+
     def test_tuple_clauselist_in(self):
         self.assert_compile(
             tuple_(table1.c.myid, table1.c.name).in_(