- The behavior of the :func:`.union` construct and related constructs

such as :meth:`.Query.union` now handle the case where the embedded
SELECT statements need to be parenthesized due to the fact that they
include LIMIT, OFFSET and/or ORDER BY.   These queries **do not work
on SQLite**, and will fail on that backend as they did before, but
should now work on all other backends.
fixes #2528

2 files changed, 83 insertions, 4 deletions

diff --git a/test/sql/test_compiler.py b/test/sql/test_compiler.py
index 06cb80ba0..7ff7d68af 100644
--- a/test/sql/test_compiler.py
+++ b/test/sql/test_compiler.py
@@ -1643,14 +1643,12 @@ class SelectTest(fixtures.TestBase, AssertsCompiledSQL):
 
         s = select([column('foo'), column('bar')])
 
-        # ORDER BY's even though not supported by
-        # all DB's, are rendered if requested
         self.assert_compile(
             union(
                 s.order_by("foo"),
                 s.order_by("bar")),
-            "SELECT foo, bar ORDER BY foo UNION SELECT foo, bar ORDER BY bar")
-        # self_group() is honored
+            "(SELECT foo, bar ORDER BY foo) UNION "
+            "(SELECT foo, bar ORDER BY bar)")
         self.assert_compile(
             union(s.order_by("foo").self_group(),
                   s.order_by("bar").limit(10).self_group()),
@@ -1759,6 +1757,67 @@ class SelectTest(fixtures.TestBase, AssertsCompiledSQL):
             "SELECT foo, bar FROM bat)"
         )
 
+        # tests for [ticket:2528]
+        # sqlite hates all of these.
+        self.assert_compile(
+            union(
+                s.limit(1),
+                s.offset(2)
+            ),
+            "(SELECT foo, bar FROM bat LIMIT :param_1) "
+            "UNION (SELECT foo, bar FROM bat LIMIT -1 OFFSET :param_2)"
+        )
+
+        self.assert_compile(
+            union(
+                s.order_by(column('bar')),
+                s.offset(2)
+            ),
+            "(SELECT foo, bar FROM bat ORDER BY bar) "
+            "UNION (SELECT foo, bar FROM bat LIMIT -1 OFFSET :param_1)"
+        )
+
+        self.assert_compile(
+            union(
+                s.limit(1).alias('a'),
+                s.limit(2).alias('b')
+            ),
+            "(SELECT foo, bar FROM bat LIMIT :param_1) "
+            "UNION (SELECT foo, bar FROM bat LIMIT :param_2)"
+        )
+
+        self.assert_compile(
+            union(
+                s.limit(1).self_group(),
+                s.limit(2).self_group()
+            ),
+            "(SELECT foo, bar FROM bat LIMIT :param_1) "
+            "UNION (SELECT foo, bar FROM bat LIMIT :param_2)"
+        )
+
+        self.assert_compile(
+            union(s.limit(1), s.limit(2).offset(3)).alias().select(),
+            "SELECT anon_1.foo, anon_1.bar FROM "
+            "((SELECT foo, bar FROM bat LIMIT :param_1) "
+            "UNION (SELECT foo, bar FROM bat LIMIT :param_2 OFFSET :param_3)) "
+            "AS anon_1"
+        )
+
+        # this version works for SQLite
+        self.assert_compile(
+            union(
+                s.limit(1).alias().select(),
+                s.offset(2).alias().select(),
+            ),
+            "SELECT anon_1.foo, anon_1.bar "
+            "FROM (SELECT foo, bar FROM bat"
+            " LIMIT :param_1) AS anon_1 "
+            "UNION SELECT anon_2.foo, anon_2.bar "
+            "FROM (SELECT foo, bar "
+            "FROM bat"
+            " LIMIT -1 OFFSET :param_2) AS anon_2"
+        )
+
     def test_binds(self):
         for (
             stmt,
diff --git a/test/sql/test_selectable.py b/test/sql/test_selectable.py
index 3390f4a77..4a332a4d1 100644
--- a/test/sql/test_selectable.py
+++ b/test/sql/test_selectable.py
@@ -458,6 +458,26 @@ class SelectableTest(
         assert u1.corresponding_column(table2.c.col1) is u1.c._all_columns[0]
         assert u1.corresponding_column(table2.c.col3) is u1.c._all_columns[2]
 
+    @testing.emits_warning("Column 'col1'")
+    def test_union_alias_dupe_keys_grouped(self):
+        s1 = select([table1.c.col1, table1.c.col2, table2.c.col1]).\
+            limit(1).alias()
+        s2 = select([table2.c.col1, table2.c.col2, table2.c.col3]).limit(1)
+        u1 = union(s1, s2)
+
+        assert u1.corresponding_column(
+            s1.c._all_columns[0]) is u1.c._all_columns[0]
+        assert u1.corresponding_column(s2.c.col1) is u1.c._all_columns[0]
+        assert u1.corresponding_column(s1.c.col2) is u1.c.col2
+        assert u1.corresponding_column(s2.c.col2) is u1.c.col2
+
+        assert u1.corresponding_column(s2.c.col3) is u1.c._all_columns[2]
+
+        # this differs from the non-alias test because table2.c.col1 is
+        # more directly at s2.c.col1 than it is s1.c.col1.
+        assert u1.corresponding_column(table2.c.col1) is u1.c._all_columns[0]
+        assert u1.corresponding_column(table2.c.col3) is u1.c._all_columns[2]
+
     def test_select_union(self):
 
         # like testaliasunion, but off a Select off the union.