implement independent CTEs

Added new method :meth:`_sql.HasCTE.add_cte` to each of the
:func:`_sql.select`, :func:`_sql.insert`, :func:`_sql.update` and
:func:`_sql.delete` constructs. This method will add the given
:class:`_sql.CTE` as an "independent" CTE of the statement, meaning it
renders in the WITH clause above the statement unconditionally even if it
is not otherwise referenced in the primary statement. This is a popular use
case on the PostgreSQL database where a CTE is used for a DML statement
that runs against database rows independently of the primary statement.

Fixes: #6752
Change-Id: Ibf635763e40269cbd10f4c17e208850d8e8d0188

1 files changed, 25 insertions, 1 deletions

diff --git a/test/dialect/postgresql/test_compiler.py b/test/dialect/postgresql/test_compiler.py
index e48de9d21..c08038df6 100644
--- a/test/dialect/postgresql/test_compiler.py
+++ b/test/dialect/postgresql/test_compiler.py
@@ -2518,7 +2518,7 @@ class InsertOnConflictTest(fixtures.TestBase, AssertsCompiledSQL):
         self.assert_compile(
             stmt,
             "WITH i_upsert AS "
-            "(INSERT INTO mytable (name) VALUES (%(name)s) "
+            "(INSERT INTO mytable (name) VALUES (%(param_1)s) "
             "ON CONFLICT (name, description) "
             "WHERE description != %(description_1)s "
             "DO UPDATE SET name = excluded.name "
@@ -2527,6 +2527,30 @@ class InsertOnConflictTest(fixtures.TestBase, AssertsCompiledSQL):
             "FROM i_upsert",
         )
 
+    def test_combined_with_cte(self):
+        t = table("t", column("c1"), column("c2"))
+
+        delete_statement_cte = t.delete().where(t.c.c1 < 1).cte("deletions")
+
+        insert_stmt = insert(t).values([{"c1": 1, "c2": 2}])
+        update_stmt = insert_stmt.on_conflict_do_update(
+            index_elements=[t.c.c1],
+            set_={
+                col.name: col
+                for col in insert_stmt.excluded
+                if col.name in ("c1", "c2")
+            },
+        ).add_cte(delete_statement_cte)
+
+        self.assert_compile(
+            update_stmt,
+            "WITH deletions AS (DELETE FROM t WHERE t.c1 < %(c1_1)s) "
+            "INSERT INTO t (c1, c2) VALUES (%(c1_m0)s, %(c2_m0)s) "
+            "ON CONFLICT (c1) DO UPDATE SET c1 = excluded.c1, "
+            "c2 = excluded.c2",
+            checkparams={"c1_m0": 1, "c2_m0": 2, "c1_1": 1},
+        )
+
     def test_quote_raw_string_col(self):
         t = table("t", column("FancyName"), column("other name"))