use the stack for insert_from_select

Fixed issue related to new ``add_cte()`` feature where pairing two
"INSERT..FROM SELECT" statements simultaneously would lose track of the two
independent SELECT statements, leading to the wrong SQL.

Fixes: #7036
Change-Id: I90fe47eb203bc5c1ea5810db0edba08250b2b7e6

diff --git a/doc/build/changelog/unreleased_14/7036.rst b/doc/build/changelog/unreleased_14/7036.rst
new file mode 100644 (file)
index 0000000..f908206
--- /dev/null
+++ b/doc/build/changelog/unreleased_14/7036.rst
@@ -0,0 +1,7 @@
+.. change::
+    :tags: bug, sql
+    :tickets: 7036
+
+    Fixed issue related to new ``add_cte()`` feature where pairing two
+    "INSERT..FROM SELECT" statements simultaneously would lose track of the two
+    independent SELECT statements, leading to the wrong SQL.

diff --git a/lib/sqlalchemy/sql/compiler.py b/lib/sqlalchemy/sql/compiler.py
index 9af82823af078ebc7813f1b3b7559ccd5880f15f..5153f54d17f62f700f43a6838f9aa12c4d439f70 100644 (file)
--- a/lib/sqlalchemy/sql/compiler.py
+++ b/lib/sqlalchemy/sql/compiler.py
@@ -3687,7 +3687,10 @@ class SQLCompiler(Compiled):
             returning_clause = None
 
         if insert_stmt.select is not None:
-            select_text = self.process(self._insert_from_select, **kw)
+            # placed here by crud.py
+            select_text = self.process(
+                self.stack[-1]["insert_from_select"], **kw
+            )
 
             if self.ctes and toplevel and self.dialect.cte_follows_insert:
                 text += " %s%s" % (self._render_cte_clause(), select_text)

diff --git a/lib/sqlalchemy/sql/crud.py b/lib/sqlalchemy/sql/crud.py
index b8f8cb4cef7638c3a9a86850dc4849b1f491727f..d43f33ebb38674a9b4eb3054384427647d77adad 100644 (file)
--- a/lib/sqlalchemy/sql/crud.py
+++ b/lib/sqlalchemy/sql/crud.py
@@ -310,7 +310,9 @@ def _scan_insert_from_select_cols(
 
     cols = [stmt.table.c[_column_as_key(name)] for name in stmt._select_names]
 
-    compiler._insert_from_select = stmt.select
+    assert compiler.stack[-1]["selectable"] is stmt
+
+    compiler.stack[-1]["insert_from_select"] = stmt.select
 
     add_select_cols = []
     if stmt.include_insert_from_select_defaults:
@@ -331,10 +333,12 @@ def _scan_insert_from_select_cols(
 
     if add_select_cols:
         values.extend(add_select_cols)
-        compiler._insert_from_select = compiler._insert_from_select._generate()
-        compiler._insert_from_select._raw_columns = tuple(
-            compiler._insert_from_select._raw_columns
+        ins_from_select = compiler.stack[-1]["insert_from_select"]
+        ins_from_select = ins_from_select._generate()
+        ins_from_select._raw_columns = tuple(
+            ins_from_select._raw_columns
         ) + tuple(expr for col, col_expr, expr in add_select_cols)
+        compiler.stack[-1]["insert_from_select"] = ins_from_select
 
 
 def _scan_cols(

diff --git a/test/sql/test_cte.py b/test/sql/test_cte.py
index cc663d53fbf300810294c53c4fd2f7b3fe56cb5d..0680472ff017be2bf8fde84f42734ac1c323b1ab 100644 (file)
--- a/test/sql/test_cte.py
+++ b/test/sql/test_cte.py
@@ -1583,6 +1583,29 @@ class CTETest(fixtures.TestBase, AssertsCompiledSQL):
             checkparams={"id": 1, "price": 20, "param_1": 10, "price_1": 50},
         )
 
+    def test_insert_from_select_uses_independent_cte(self):
+        """test #7036"""
+
+        t1 = table("table1", column("id1"), column("a"))
+
+        t2 = table("table2", column("id2"), column("b"))
+
+        ins1 = t1.insert().from_select(["id1", "a"], select(1, text("'a'")))
+
+        cte1 = ins1.cte("cte1")
+
+        ins2 = t2.insert().from_select(["id2", "b"], select(2, text("'b'")))
+
+        ins2 = ins2.add_cte(cte1)
+
+        self.assert_compile(
+            ins2,
+            "WITH cte1 AS "
+            "(INSERT INTO table1 (id1, a) SELECT 1, 'a') "
+            "INSERT INTO table2 (id2, b) SELECT 2, 'b'",
+            checkparams={},
+        )
+
     def test_update_uses_independent_cte(self):
         products = table("products", column("id"), column("price"))