Run PK/FK sync for multi-level inheritance w/ no intermediary update

Also fix DetectKeySwitch for intermediary class relationship

Fixed a series of related bugs regarding joined table inheritance more than
two levels deep, in conjunction with modification to primary key values,
where those primary key columns are also linked together in a foreign key
relationship as is typical for joined table inheritance.  The intermediary
table in a  three-level inheritance hierachy will now get its UPDATE if
only the primary key value has changed and passive_updates=False (e.g.
foreign key constraints not being enforced), whereas before it would be
skipped; similarly, with passive_updates=True (e.g. ON UPDATE  CASCADE in
effect), the third-level table will not receive an UPDATE statement as was
the case earlier which would fail since CASCADE already modified it.   In a
related issue, a relationship linked to a three-level inheritance hierarchy
on the primary key of an intermediary table of a joined-inheritance
hierarchy will also correctly have its foreign key column updated when the
parent object's primary key is modified, even if that parent object is a
subclass of the linked parent class, whereas before these classes would
not be counted.

Fixes: #4723
Change-Id: Idc408ead67702068e64d583a15149dd4beeefc24

4 files changed, 213 insertions, 53 deletions

diff --git a/doc/build/changelog/unreleased_13/4723.rst b/doc/build/changelog/unreleased_13/4723.rst
new file mode 100644
index 000000000..3fb4957a3
--- /dev/null
+++ b/doc/build/changelog/unreleased_13/4723.rst
@@ -0,0 +1,20 @@
+.. change::
+    :tags: bug, orm
+    :tickets: 4723
+
+    Fixed a series of related bugs regarding joined table inheritance more than
+    two levels deep, in conjunction with modification to primary key values,
+    where those primary key columns are also linked together in a foreign key
+    relationship as is typical for joined table inheritance.  The intermediary
+    table in a  three-level inheritance hierachy will now get its UPDATE if
+    only the primary key value has changed and passive_updates=False (e.g.
+    foreign key constraints not being enforced), whereas before it would be
+    skipped; similarly, with passive_updates=True (e.g. ON UPDATE  CASCADE in
+    effect), the third-level table will not receive an UPDATE statement as was
+    the case earlier which would fail since CASCADE already modified it.   In a
+    related issue, a relationship linked to a three-level inheritance hierarchy
+    on the primary key of an intermediary table of a joined-inheritance
+    hierarchy will also correctly have its foreign key column updated when the
+    parent object's primary key is modified, even if that parent object is a
+    subclass of the linked parent class, whereas before these classes would
+    not be counted.
diff --git a/lib/sqlalchemy/orm/dependency.py b/lib/sqlalchemy/orm/dependency.py
index 8a77d5052..f0ba6051d 100644
--- a/lib/sqlalchemy/orm/dependency.py
+++ b/lib/sqlalchemy/orm/dependency.py
@@ -622,7 +622,8 @@ class OneToManyDP(DependencyProcessor):
 class ManyToOneDP(DependencyProcessor):
     def __init__(self, prop):
         DependencyProcessor.__init__(self, prop)
-        self.mapper._dependency_processors.append(DetectKeySwitch(prop))
+        for mapper in self.mapper.self_and_descendants:
+            mapper._dependency_processors.append(DetectKeySwitch(prop))
 
     def per_property_dependencies(
         self,
diff --git a/lib/sqlalchemy/orm/persistence.py b/lib/sqlalchemy/orm/persistence.py
index 5106bff94..072d34f8c 100644
--- a/lib/sqlalchemy/orm/persistence.py
+++ b/lib/sqlalchemy/orm/persistence.py
@@ -680,6 +680,7 @@ def _collect_update_commands(
             continue
 
         has_all_pks = True
+        expect_pk_cascaded = False
         if bulk:
             # keys here are mapped attribute keys, so
             # look at mapper attribute keys for pk
@@ -704,6 +705,7 @@ def _collect_update_commands(
                         or ("pk_cascaded", state, col)
                         in uowtransaction.attributes
                     ):
+                        expect_pk_cascaded = True
                         pk_params[col._label] = history.added[0]
                         params.pop(col.key, None)
                     else:
@@ -731,6 +733,22 @@ def _collect_update_commands(
                 has_all_defaults,
                 has_all_pks,
             )
+        elif expect_pk_cascaded:
+            # no UPDATE occurs on this table, but we expect that CASCADE rules
+            # have changed the primary key of the row; propagate this event to
+            # other columns that expect to have been modified. this normally
+            # occurs after the UPDATE is emitted however we invoke it here
+            # explicitly in the absense of our invoking an UPDATE
+            for m, equated_pairs in mapper._table_to_equated[table]:
+                sync.populate(
+                    state,
+                    m,
+                    state,
+                    m,
+                    equated_pairs,
+                    uowtransaction,
+                    mapper.passive_updates,
+                )
 
 
 def _collect_post_update_commands(
diff --git a/test/orm/test_naturalpks.py b/test/orm/test_naturalpks.py
index 982f34241..3788eb329 100644
--- a/test/orm/test_naturalpks.py
+++ b/test/orm/test_naturalpks.py
@@ -1463,6 +1463,19 @@ class JoinedInheritanceTest(fixtures.MappedTest):
             test_needs_fk=True,
         )
 
+        Table(
+            "owner",
+            metadata,
+            Column(
+                "name",
+                String(50),
+                ForeignKey("manager.name", **fk_args),
+                primary_key=True,
+            ),
+            Column("owner_name", String(50)),
+            test_needs_fk=True,
+        )
+
     @classmethod
     def setup_classes(cls):
         class Person(cls.Comparable):
@@ -1474,31 +1487,15 @@ class JoinedInheritanceTest(fixtures.MappedTest):
         class Manager(Person):
             pass
 
-    @testing.requires.on_update_cascade
-    def test_pk_passive(self):
-        self._test_pk(True)
-
-    @testing.requires.non_updating_cascade
-    def test_pk_nonpassive(self):
-        self._test_pk(False)
-
-    @testing.requires.on_update_cascade
-    def test_fk_passive(self):
-        self._test_fk(True)
-
-    # PG etc. need passive=True to allow PK->PK cascade
-    @testing.requires.non_updating_cascade
-    def test_fk_nonpassive(self):
-        self._test_fk(False)
+        class Owner(Manager):
+            pass
 
-    def _test_pk(self, passive_updates):
-        Person, Manager, person, manager, Engineer, engineer = (
-            self.classes.Person,
-            self.classes.Manager,
-            self.tables.person,
-            self.tables.manager,
-            self.classes.Engineer,
-            self.tables.engineer,
+    def _mapping_fixture(self, threelevel, passive_updates):
+        Person, Manager, Engineer, Owner = self.classes(
+            "Person", "Manager", "Engineer", "Owner"
+        )
+        person, manager, engineer, owner = self.tables(
+            "person", "manager", "engineer", "owner"
         )
 
         mapper(
@@ -1508,6 +1505,7 @@ class JoinedInheritanceTest(fixtures.MappedTest):
             polymorphic_identity="person",
             passive_updates=passive_updates,
         )
+
         mapper(
             Engineer,
             engineer,
@@ -1521,10 +1519,53 @@ class JoinedInheritanceTest(fixtures.MappedTest):
                 )
             },
         )
+
         mapper(
             Manager, manager, inherits=Person, polymorphic_identity="manager"
         )
 
+        if threelevel:
+            mapper(
+                Owner, owner, inherits=Manager, polymorphic_identity="owner"
+            )
+
+    @testing.requires.on_update_cascade
+    def test_pk_passive(self):
+        self._test_pk(True)
+
+    @testing.requires.non_updating_cascade
+    def test_pk_nonpassive(self):
+        self._test_pk(False)
+
+    @testing.requires.on_update_cascade
+    def test_fk_passive(self):
+        self._test_fk(True)
+
+    # PG etc. need passive=True to allow PK->PK cascade
+    @testing.requires.non_updating_cascade
+    def test_fk_nonpassive(self):
+        self._test_fk(False)
+
+    @testing.requires.on_update_cascade
+    def test_pk_threelevel_passive(self):
+        self._test_pk_threelevel(True)
+
+    @testing.requires.non_updating_cascade
+    def test_pk_threelevel_nonpassive(self):
+        self._test_pk_threelevel(False)
+
+    @testing.requires.on_update_cascade
+    def test_fk_threelevel_passive(self):
+        self._test_fk_threelevel(True)
+
+    # PG etc. need passive=True to allow PK->PK cascade
+    @testing.requires.non_updating_cascade
+    def test_fk_threelevel_nonpassive(self):
+        self._test_fk_threelevel(False)
+
+    def _test_pk(self, passive_updates):
+        Engineer, = self.classes("Engineer")
+        self._mapping_fixture(False, passive_updates)
         sess = sa.orm.sessionmaker()()
 
         e1 = Engineer(name="dilbert", primary_language="java")
@@ -1532,45 +1573,115 @@ class JoinedInheritanceTest(fixtures.MappedTest):
         sess.commit()
         e1.name = "wally"
         e1.primary_language = "c++"
+
         sess.commit()
+        eq_(
+            sess.execute(self.tables.engineer.select()).fetchall(),
+            [("wally", "c++", None)],
+        )
+
+        eq_(e1.name, "wally")
+
+        e1.name = "dogbert"
+        sess.commit()
+        eq_(e1.name, "dogbert")
+
+        eq_(
+            sess.execute(self.tables.engineer.select()).fetchall(),
+            [("dogbert", "c++", None)],
+        )
 
     def _test_fk(self, passive_updates):
-        Person, Manager, person, manager, Engineer, engineer = (
-            self.classes.Person,
-            self.classes.Manager,
-            self.tables.person,
-            self.tables.manager,
-            self.classes.Engineer,
-            self.tables.engineer,
+        Manager, Engineer = self.classes("Manager", "Engineer")
+
+        self._mapping_fixture(False, passive_updates)
+
+        sess = sa.orm.sessionmaker()()
+
+        m1 = Manager(name="dogbert", paperwork="lots")
+        e1, e2 = (
+            Engineer(name="dilbert", primary_language="java", boss=m1),
+            Engineer(name="wally", primary_language="c++", boss=m1),
         )
+        sess.add_all([e1, e2, m1])
+        sess.commit()
 
-        mapper(
-            Person,
-            person,
-            polymorphic_on=person.c.type,
-            polymorphic_identity="person",
-            passive_updates=passive_updates,
+        eq_(e1.boss_name, "dogbert")
+        eq_(e2.boss_name, "dogbert")
+
+        eq_(
+            sess.execute(
+                self.tables.engineer.select().order_by(Engineer.name)
+            ).fetchall(),
+            [("dilbert", "java", "dogbert"), ("wally", "c++", "dogbert")],
         )
-        mapper(
-            Engineer,
-            engineer,
-            inherits=Person,
-            polymorphic_identity="engineer",
-            properties={
-                "boss": relationship(
-                    Manager,
-                    primaryjoin=manager.c.name == engineer.c.boss_name,
-                    passive_updates=passive_updates,
-                )
-            },
+
+        sess.expire_all()
+
+        m1.name = "pointy haired"
+        e1.primary_language = "scala"
+        e2.primary_language = "cobol"
+        sess.commit()
+
+        eq_(e1.boss_name, "pointy haired")
+        eq_(e2.boss_name, "pointy haired")
+
+        eq_(
+            sess.execute(
+                self.tables.engineer.select().order_by(Engineer.name)
+            ).fetchall(),
+            [
+                ("dilbert", "scala", "pointy haired"),
+                ("wally", "cobol", "pointy haired"),
+            ],
         )
-        mapper(
-            Manager, manager, inherits=Person, polymorphic_identity="manager"
+
+    def _test_pk_threelevel(self, passive_updates):
+        Owner, = self.classes("Owner")
+
+        self._mapping_fixture(True, passive_updates)
+
+        sess = sa.orm.sessionmaker()()
+
+        o1 = Owner(name="dogbert", owner_name="dog")
+        sess.add(o1)
+        sess.commit()
+        o1.name = "pointy haired"
+        o1.owner_name = "pointy"
+        sess.commit()
+
+        eq_(
+            sess.execute(self.tables.manager.select()).fetchall(),
+            [("pointy haired", None)],
         )
+        eq_(
+            sess.execute(self.tables.owner.select()).fetchall(),
+            [("pointy haired", "pointy")],
+        )
+
+        eq_(o1.name, "pointy haired")
+
+        o1.name = "catbert"
+        sess.commit()
+
+        eq_(o1.name, "catbert")
+
+        eq_(
+            sess.execute(self.tables.manager.select()).fetchall(),
+            [("catbert", None)],
+        )
+        eq_(
+            sess.execute(self.tables.owner.select()).fetchall(),
+            [("catbert", "pointy")],
+        )
+
+    def _test_fk_threelevel(self, passive_updates):
+        Owner, Engineer = self.classes("Owner", "Engineer")
+        self._mapping_fixture(True, passive_updates)
 
         sess = sa.orm.sessionmaker()()
 
-        m1 = Manager(name="dogbert", paperwork="lots")
+        m1 = Owner(name="dogbert", paperwork="lots", owner_name="dog")
         e1, e2 = (
             Engineer(name="dilbert", primary_language="java", boss=m1),
             Engineer(name="wally", primary_language="c++", boss=m1),
@@ -1583,6 +1694,7 @@ class JoinedInheritanceTest(fixtures.MappedTest):
         sess.expire_all()
 
         m1.name = "pointy haired"
+
         e1.primary_language = "scala"
         e2.primary_language = "cobol"
         sess.commit()
@@ -1590,6 +1702,15 @@ class JoinedInheritanceTest(fixtures.MappedTest):
         eq_(e1.boss_name, "pointy haired")
         eq_(e2.boss_name, "pointy haired")
 
+        eq_(
+            sess.execute(self.tables.manager.select()).fetchall(),
+            [("pointy haired", "lots")],
+        )
+        eq_(
+            sess.execute(self.tables.owner.select()).fetchall(),
+            [("pointy haired", "dog")],
+        )
+
 
 class JoinedInheritancePKOnFKTest(fixtures.MappedTest):
     """Test cascades of pk->non-pk/fk on joined table inh."""