diff options
| author | Eric Masseran <eric.masseran@gmail.com> | 2021-11-02 16:40:04 -0400 |
|---|---|---|
| committer | Mike Bayer <mike_mp@zzzcomputing.com> | 2021-11-16 14:44:36 -0500 |
| commit | 958f902b1fc528fed0be550bc573545de47ed854 (patch) | |
| tree | 806989f5c2fe6f3a2480b6f878aa11db0eed92ca /test/sql | |
| parent | 6206f0ff74e95c9339dc0f0e26caab55e9bcda45 (diff) | |
| download | sqlalchemy-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.py | 47 | ||||
| -rw-r--r-- | test/sql/test_select.py | 23 |
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}, + ) |
