allow CTE to be direct DML target

Implemented support for a :class:`_sql.CTE` construct to be used directly
as the target of a :func:`_sql.delete` construct, i.e. "WITH ... AS cte
DELETE FROM cte". This appears to be a useful feature of SQL Server.

The CTE is now generally usable as a DML target table however
it's not clear if this syntax is valid beyond the use case of
DELETE itself.

Fixes: #6464
Change-Id: I3aac6fae2a1abb39bc0ffa87a044f5eb4f90f026

diff --git a/doc/build/changelog/unreleased_14/6464.rst b/doc/build/changelog/unreleased_14/6464.rst
new file mode 100644 (file)
index 0000000..8011841
--- /dev/null
+++ b/doc/build/changelog/unreleased_14/6464.rst
@@ -0,0 +1,7 @@
+.. change::
+    :tags: usecase, mssql
+    :tickets: 6464
+
+    Implemented support for a :class:`_sql.CTE` construct to be used directly
+    as the target of a :func:`_sql.delete` construct, i.e. "WITH ... AS cte
+    DELETE FROM cte". This appears to be a useful feature of SQL Server.

diff --git a/lib/sqlalchemy/sql/selectable.py b/lib/sqlalchemy/sql/selectable.py
index 43ba0da4cbd91483dfdf53fb12ed08d9555c2ed6..7007bb430e70c997d2e2b930973f9a8d6e5d7cdc 100644 (file)
--- a/lib/sqlalchemy/sql/selectable.py
+++ b/lib/sqlalchemy/sql/selectable.py
@@ -1992,7 +1992,13 @@ class TableSample(AliasedReturnsRows):
             return functions.func.system(self.sampling)
 
 
-class CTE(Generative, HasPrefixes, HasSuffixes, AliasedReturnsRows):
+class CTE(
+    roles.DMLTableRole,
+    Generative,
+    HasPrefixes,
+    HasSuffixes,
+    AliasedReturnsRows,
+):
     """Represent a Common Table Expression.
 
     The :class:`_expression.CTE` object is obtained using the

diff --git a/test/sql/test_cte.py b/test/sql/test_cte.py
index 7752a9b815559688797f971a6a999f304cab9daf..25d7e33ba03f365ceea21af5e6f140ec071ee836 100644 (file)
--- a/test/sql/test_cte.py
+++ b/test/sql/test_cte.py
@@ -1,4 +1,7 @@
+from sqlalchemy import delete
 from sqlalchemy import testing
+from sqlalchemy import text
+from sqlalchemy import update
 from sqlalchemy.dialects import mssql
 from sqlalchemy.engine import default
 from sqlalchemy.exc import CompileError
@@ -1231,6 +1234,76 @@ class CTETest(fixtures.TestBase, AssertsCompiledSQL):
         )
         eq_(stmt.compile().isupdate, True)
 
+    def test_update_against_cte_directly(self):
+        """test #6464
+
+        for UPDATE, I'm not sure this is a valid syntax on any platform.
+
+        """
+        products = table("products", column("id"), column("price"))
+
+        cte = products.select().cte("pd")
+        assert "autocommit" not in cte.select()._execution_options
+
+        stmt = update(cte)
+        eq_(stmt.compile().execution_options["autocommit"], True)
+
+        self.assert_compile(
+            stmt,
+            "WITH pd AS (SELECT products.id AS id, products.price AS price "
+            "FROM products) UPDATE pd SET id=:id, price=:price",
+        )
+        eq_(stmt.compile().isupdate, True)
+
+    def test_delete_against_cte_directly(self):
+        """test #6464.
+
+        SQL-Server specific arrangement seems to allow
+        DELETE from a CTE directly.
+
+        """
+        products = table("products", column("id"), column("price"))
+
+        cte = products.select().cte("pd")
+        assert "autocommit" not in cte.select()._execution_options
+
+        stmt = delete(cte)
+        eq_(stmt.compile().execution_options["autocommit"], True)
+
+        self.assert_compile(
+            stmt,
+            "WITH pd AS (SELECT products.id AS id, products.price AS price "
+            "FROM products) DELETE FROM pd",
+        )
+        eq_(stmt.compile().isdelete, True)
+
+    def test_delete_against_user_textual_cte(self):
+        """test #6464.
+
+        Test the user's exact arrangement.
+
+        """
+
+        q = select(
+            text(
+                "name, date_hour, "
+                "ROW_NUMBER() OVER(PARTITION BY name, date_hour "
+                "ORDER BY value DESC)"
+                " AS RN FROM testtable"
+            )
+        )
+        cte = q.cte("deldup")
+        stmt = delete(cte, text("RN > 1"))
+        eq_(stmt.compile().execution_options["autocommit"], True)
+
+        self.assert_compile(
+            stmt,
+            "WITH deldup AS (SELECT name, date_hour, ROW_NUMBER() "
+            "OVER(PARTITION BY name, date_hour ORDER BY value DESC) "
+            "AS RN FROM testtable) DELETE FROM deldup WHERE RN > 1",
+        )
+        eq_(stmt.compile().isdelete, True)
+
     def test_standalone_function(self):
         a = table("a", column("x"))
         a_stmt = select(a)