Merge "ensure implicit_returning is checked on a Table instance" into main

3 files changed, 73 insertions, 1 deletions

diff --git a/doc/build/changelog/unreleased_20/8812.rst b/doc/build/changelog/unreleased_20/8812.rst
new file mode 100644
index 000000000..a354ce0f9
--- /dev/null
+++ b/doc/build/changelog/unreleased_20/8812.rst
@@ -0,0 +1,8 @@
+.. change::
+    :tags: bug, orm, regression
+    :tickets: 8812
+
+    Fixed regression where flushing a mapped class that's mapped against a
+    subquery, such as a direct mapping or some forms of concrete table
+    inheritance, would fail if the :paramref:`_orm.Mapper.eager_defaults`
+    parameter were used.
diff --git a/lib/sqlalchemy/orm/persistence.py b/lib/sqlalchemy/orm/persistence.py
index 77532f323..c236ad1cf 100644
--- a/lib/sqlalchemy/orm/persistence.py
+++ b/lib/sqlalchemy/orm/persistence.py
@@ -992,7 +992,7 @@ def _emit_insert_statements(
                 or (
                     has_all_defaults
                     or not base_mapper.eager_defaults
-                    or not base_mapper.local_table.implicit_returning
+                    or not table.implicit_returning
                     or not connection.dialect.insert_returning
                 )
             )
diff --git a/test/orm/test_unitofworkv2.py b/test/orm/test_unitofworkv2.py
index 0fac7ccda..f204e954c 100644
--- a/test/orm/test_unitofworkv2.py
+++ b/test/orm/test_unitofworkv2.py
@@ -3046,6 +3046,70 @@ class EagerDefaultsTest(fixtures.MappedTest):
             ),
         )
 
+    @testing.fixture
+    def selectable_fixture(self, decl_base):
+
+        t1, t2 = self.tables("test", "test2")
+
+        stmt = (
+            select(t1.c.id, t1.c.foo, t2.c.id.label("id2"), t2.c.bar)
+            .join_from(t1, t2, t1.c.foo == t2.c.foo)
+            .subquery()
+        )
+
+        class MyClass(decl_base):
+            __table__ = stmt
+
+            __mapper_args__ = {"eager_defaults": True}
+
+        return MyClass
+
+    def test_against_selectable_insert(self, selectable_fixture):
+        """test #8812"""
+        MyClass = selectable_fixture
+
+        s = fixture_session()
+        obj = MyClass(id=1, id2=1, bar=5)
+        s.add(obj)
+
+        with self.sql_execution_asserter() as asserter:
+            s.flush()
+
+        asserter.assert_(
+            Conditional(
+                testing.db.dialect.insert_executemany_returning,
+                [
+                    CompiledSQL(
+                        "INSERT INTO test (id) VALUES (:id) "
+                        "RETURNING test.foo",
+                        [{"id": 1}],
+                    ),
+                    CompiledSQL(
+                        "INSERT INTO test2 (id, bar) VALUES (:id, :bar)",
+                        [{"id": 1, "bar": 5}],
+                    ),
+                ],
+                [
+                    CompiledSQL(
+                        "INSERT INTO test (id) VALUES (:id)",
+                        [{"id": 1}],
+                    ),
+                    CompiledSQL(
+                        "INSERT INTO test2 (id, bar) VALUES (:id, :bar)",
+                        [{"id": 1, "bar": 5}],
+                    ),
+                    CompiledSQL(
+                        "SELECT anon_1.foo AS anon_1_foo FROM "
+                        "(SELECT test.id AS id, test.foo AS foo, "
+                        "test2.id AS id2, test2.bar AS bar FROM test "
+                        "JOIN test2 ON test.foo = test2.foo) AS anon_1 "
+                        "WHERE anon_1.id = :pk_1 AND anon_1.id2 = :pk_2",
+                        [{"pk_1": 1, "pk_2": 1}],
+                    ),
+                ],
+            ),
+        )
+
 
 class TypeWoBoolTest(fixtures.MappedTest, testing.AssertsExecutionResults):
     """test support for custom datatypes that return a non-__bool__ value