raise informative error when selectable can't be extended

An informative error is raised for the use case where
:meth:`.Insert.from_select` is being passed a "compound select" object such
as a UNION, yet the INSERT statement needs to append additional columns to
support Python-side or explicit SQL defaults from the table metadata. In
this case a subquery of the compound object should be passed.

Fixes: #8073
Change-Id: Ic4a5dbf84ec49d2451901be05cb9cf6ae93f02b7

diff --git a/doc/build/changelog/unreleased_14/8073.rst b/doc/build/changelog/unreleased_14/8073.rst
new file mode 100644 (file)
index 0000000..57add15
--- /dev/null
+++ b/doc/build/changelog/unreleased_14/8073.rst
@@ -0,0 +1,9 @@
+.. change::
+    :tags: bug, sql
+    :tickets: 8073
+
+    An informative error is raised for the use case where
+    :meth:`.Insert.from_select` is being passed a "compound select" object such
+    as a UNION, yet the INSERT statement needs to append additional columns to
+    support Python-side or explicit SQL defaults from the table metadata. In
+    this case a subquery of the compound object should be passed.

diff --git a/lib/sqlalchemy/sql/crud.py b/lib/sqlalchemy/sql/crud.py
index b0cb3cbf826ccdd7a10ac333b1a98337f1e71411..913e4d43338579c9abf29ed75f538182657847ea 100644 (file)
--- a/lib/sqlalchemy/sql/crud.py
+++ b/lib/sqlalchemy/sql/crud.py
@@ -35,6 +35,7 @@ from . import roles
 from .elements import ColumnClause
 from .schema import default_is_clause_element
 from .schema import default_is_sequence
+from .selectable import Select
 from .selectable import TableClause
 from .. import exc
 from .. import util
@@ -486,10 +487,21 @@ def _scan_insert_from_select_cols(
     if add_select_cols:
         values.extend(add_select_cols)
         ins_from_select = compiler.stack[-1]["insert_from_select"]
+        if not isinstance(ins_from_select, Select):
+            raise exc.CompileError(
+                f"Can't extend statement for INSERT..FROM SELECT to include "
+                f"additional default-holding column(s) "
+                f"""{
+                    ', '.join(repr(key) for _, key, _ in add_select_cols)
+                }.  Convert the selectable to a subquery() first, or pass """
+                "include_defaults=False to Insert.from_select() to skip these "
+                "columns."
+            )
         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)
+        # copy raw_columns
+        ins_from_select._raw_columns = list(ins_from_select._raw_columns) + [
+            expr for col, col_expr, expr in add_select_cols
+        ]
         compiler.stack[-1]["insert_from_select"] = ins_from_select
 
 

diff --git a/lib/sqlalchemy/sql/dml.py b/lib/sqlalchemy/sql/dml.py
index f063cf78169e9c7847b30e52d1685b866184f22f..e63a34454d68a856b527504e0bea8b31dd9dddde 100644 (file)
--- a/lib/sqlalchemy/sql/dml.py
+++ b/lib/sqlalchemy/sql/dml.py
@@ -87,6 +87,7 @@ if TYPE_CHECKING:
     from .selectable import _ColumnsClauseElement
     from .selectable import _SelectIterable
     from .selectable import Select
+    from .selectable import Selectable
 
     def isupdate(dml: DMLState) -> TypeGuard[UpdateDMLState]:
         ...
@@ -1040,7 +1041,7 @@ class Insert(ValuesBase):
     def from_select(
         self: SelfInsert,
         names: List[str],
-        select: Select[Any],
+        select: Selectable,
         include_defaults: bool = True,
     ) -> SelfInsert:
         """Return a new :class:`_expression.Insert` construct which represents

diff --git a/test/sql/test_insert.py b/test/sql/test_insert.py
index 2f9f9a4f7688fb09b3329ba72306ebb82fa90b56..3a6217f6712be6294cec4462565a0c331a4cbb17 100644 (file)
--- a/test/sql/test_insert.py
+++ b/test/sql/test_insert.py
@@ -24,6 +24,7 @@ from sqlalchemy.testing import assert_raises
 from sqlalchemy.testing import assert_raises_message
 from sqlalchemy.testing import AssertsCompiledSQL
 from sqlalchemy.testing import eq_
+from sqlalchemy.testing import expect_raises_message
 from sqlalchemy.testing import expect_warnings
 from sqlalchemy.testing import fixtures
 
@@ -662,6 +663,75 @@ class InsertTest(_InsertTestBase, fixtures.TablesTest, AssertsCompiledSQL):
             checkparams={"name_1": "foo", "foo": None},
         )
 
+    def test_insert_from_select_fn_defaults_compound(self):
+        """test #8073"""
+
+        metadata = MetaData()
+
+        table = Table(
+            "sometable",
+            metadata,
+            Column("id", Integer, primary_key=True),
+            Column("foo", Integer, default="foo"),
+            Column("bar", Integer, default="bar"),
+        )
+        table1 = self.tables.mytable
+        sel = (
+            select(table1.c.myid)
+            .where(table1.c.name == "foo")
+            .union(select(table1.c.myid).where(table1.c.name == "foo"))
+        )
+        ins = table.insert().from_select(["id"], sel)
+        with expect_raises_message(
+            exc.CompileError,
+            r"Can't extend statement for INSERT..FROM SELECT to include "
+            r"additional default-holding column\(s\) 'foo', 'bar'.  "
+            r"Convert the selectable to a subquery\(\) first, or pass "
+            r"include_defaults=False to Insert.from_select\(\) to skip these "
+            r"columns.",
+        ):
+            ins.compile()
+
+    def test_insert_from_select_fn_defaults_compound_subquery(self):
+        """test #8073"""
+
+        metadata = MetaData()
+
+        def foo(ctx):
+            return 12
+
+        table = Table(
+            "sometable",
+            metadata,
+            Column("id", Integer, primary_key=True),
+            Column("foo", Integer, default="foo"),
+            Column("bar", Integer, default="bar"),
+        )
+        table1 = self.tables.mytable
+        sel = (
+            select(table1.c.myid)
+            .where(table1.c.name == "foo")
+            .union(select(table1.c.myid).where(table1.c.name == "foo"))
+            .subquery()
+        )
+
+        ins = table.insert().from_select(["id"], sel)
+        self.assert_compile(
+            ins,
+            "INSERT INTO sometable (id, foo, bar) SELECT anon_1.myid, "
+            ":foo AS anon_2, :bar AS anon_3 FROM "
+            "(SELECT mytable.myid AS myid FROM mytable "
+            "WHERE mytable.name = :name_1 UNION "
+            "SELECT mytable.myid AS myid FROM mytable "
+            "WHERE mytable.name = :name_2) AS anon_1",
+            checkparams={
+                "foo": None,
+                "bar": None,
+                "name_1": "foo",
+                "name_2": "foo",
+            },
+        )
+
     def test_insert_from_select_dont_mutate_raw_columns(self):
         # test [ticket:3603]
         from sqlalchemy import table