summaryrefslogtreecommitdiff
path: root/test/sql
diff options
context:
space:
mode:
authorEric Masseran <eric.masseran@gmail.com>2021-11-02 16:40:04 -0400
committerMike Bayer <mike_mp@zzzcomputing.com>2021-11-16 14:44:36 -0500
commit958f902b1fc528fed0be550bc573545de47ed854 (patch)
tree806989f5c2fe6f3a2480b6f878aa11db0eed92ca /test/sql
parent6206f0ff74e95c9339dc0f0e26caab55e9bcda45 (diff)
downloadsqlalchemy-958f902b1fc528fed0be550bc573545de47ed854.tar.gz
Add Non linear CTE support
"Compound select" methods like :meth:`_sql.Select.union`, :meth:`_sql.Select.intersect_all` etc. now accept ``*other`` as an argument rather than ``other`` to allow for multiple additional SELECTs to be compounded with the parent statement at once. In particular, the change as applied to :meth:`_sql.CTE.union` and :meth:`_sql.CTE.union_all` now allow for a so-called "non-linear CTE" to be created with the :class:`_sql.CTE` construct, whereas previously there was no way to have more than two CTE sub-elements in a UNION together while still correctly calling upon the CTE in recursive fashion. Pull request courtesy Eric Masseran. Allow: ```sql WITH RECURSIVE nodes(x) AS ( SELECT 59 UNION SELECT aa FROM edge JOIN nodes ON bb=x UNION SELECT bb FROM edge JOIN nodes ON aa=x ) SELECT x FROM nodes; ``` Based on @zzzeek suggestion: https://github.com/sqlalchemy/sqlalchemy/pull/7133#issuecomment-933882348 Fixes: #7259 Closes: #7260 Pull-request: https://github.com/sqlalchemy/sqlalchemy/pull/7260 Pull-request-sha: 2565a5fd4b1940e92125e53aeaa731cc682f49bb Change-Id: I685c8379762b5fb6ab4107ff8f4d8a4de70c0ca6
Diffstat (limited to 'test/sql')
-rw-r--r--test/sql/test_cte.py47
-rw-r--r--test/sql/test_select.py23
2 files changed, 69 insertions, 1 deletions
diff --git a/test/sql/test_cte.py b/test/sql/test_cte.py
index 4a34a5b04..9b0add713 100644
--- a/test/sql/test_cte.py
+++ b/test/sql/test_cte.py
@@ -1757,6 +1757,53 @@ class CTETest(fixtures.TestBase, AssertsCompiledSQL):
"foo",
)
+ def test_recursive_cte_with_multiple_union(self):
+ root_query = select(literal(1).label("val")).cte(
+ "increasing", recursive=True
+ )
+ rec_part_1 = select((root_query.c.val + 3).label("val")).where(
+ root_query.c.val < 15
+ )
+ rec_part_2 = select((root_query.c.val + 5).label("val")).where(
+ root_query.c.val < 15
+ )
+ union_rec_query = root_query.union(rec_part_1, rec_part_2)
+ union_stmt = select(union_rec_query)
+ self.assert_compile(
+ union_stmt,
+ "WITH RECURSIVE increasing(val) AS "
+ "(SELECT :param_1 AS val "
+ "UNION SELECT increasing.val + :val_1 AS val FROM increasing "
+ "WHERE increasing.val < :val_2 "
+ "UNION SELECT increasing.val + :val_3 AS val FROM increasing "
+ "WHERE increasing.val < :val_4) "
+ "SELECT increasing.val FROM increasing",
+ )
+
+ def test_recursive_cte_with_multiple_union_all(self):
+ root_query = select(literal(1).label("val")).cte(
+ "increasing", recursive=True
+ )
+ rec_part_1 = select((root_query.c.val + 3).label("val")).where(
+ root_query.c.val < 15
+ )
+ rec_part_2 = select((root_query.c.val + 5).label("val")).where(
+ root_query.c.val < 15
+ )
+
+ union_all_rec_query = root_query.union_all(rec_part_1, rec_part_2)
+ union_all_stmt = select(union_all_rec_query)
+ self.assert_compile(
+ union_all_stmt,
+ "WITH RECURSIVE increasing(val) AS "
+ "(SELECT :param_1 AS val "
+ "UNION ALL SELECT increasing.val + :val_1 AS val FROM increasing "
+ "WHERE increasing.val < :val_2 "
+ "UNION ALL SELECT increasing.val + :val_3 AS val FROM increasing "
+ "WHERE increasing.val < :val_4) "
+ "SELECT increasing.val FROM increasing",
+ )
+
class NestingCTETest(fixtures.TestBase, AssertsCompiledSQL):
diff --git a/test/sql/test_select.py b/test/sql/test_select.py
index 17b47d96d..c9abb7fb8 100644
--- a/test/sql/test_select.py
+++ b/test/sql/test_select.py
@@ -8,15 +8,16 @@ from sqlalchemy import MetaData
from sqlalchemy import select
from sqlalchemy import String
from sqlalchemy import Table
+from sqlalchemy import testing
from sqlalchemy import tuple_
from sqlalchemy import union
from sqlalchemy.sql import column
+from sqlalchemy.sql import literal
from sqlalchemy.sql import table
from sqlalchemy.testing import assert_raises_message
from sqlalchemy.testing import AssertsCompiledSQL
from sqlalchemy.testing import fixtures
-
table1 = table(
"mytable",
column("myid", Integer),
@@ -412,3 +413,23 @@ class FutureSelectTest(fixtures.TestBase, AssertsCompiledSQL):
"SELECT anon_1.name FROM (SELECT mytable.name AS name, "
"(mytable.myid, mytable.name) AS anon_2 FROM mytable) AS anon_1",
)
+
+ @testing.combinations(
+ ("union_all", "UNION ALL"),
+ ("union", "UNION"),
+ ("intersect_all", "INTERSECT ALL"),
+ ("intersect", "INTERSECT"),
+ ("except_all", "EXCEPT ALL"),
+ ("except_", "EXCEPT"),
+ )
+ def test_select_multiple_compound_elements(self, methname, joiner):
+ stmt = select(literal(1))
+ meth = getattr(stmt, methname)
+ stmt = meth(select(literal(2)), select(literal(3)))
+
+ self.assert_compile(
+ stmt,
+ "SELECT :param_1 AS anon_1"
+ " %(joiner)s SELECT :param_2 AS anon_2"
+ " %(joiner)s SELECT :param_3 AS anon_3" % {"joiner": joiner},
+ )