Improve check for overlapping FK targets on sibling classes

Fixed bug where ORM relationship would warn against conflicting sync
targets (e.g. two relationships would both write to the same column) for
sibling classes in an inheritance hierarchy, where the two relationships
would never actually conflict during writes.

Change-Id: I9367a7978cadc59066e89fc4917d7eb6c78dedee
Fixes: #4078
(cherry picked from commit 8ba8dd23b7)

doc/build/changelog/unreleased_11/4078.rst

@@ -0,0 +1,9 @@
+.. change::
+    :tags: bug, orm
+    :tickets: 4078
+    :versions: 1.2.0b3
+
+    Fixed bug where ORM relationship would warn against conflicting sync
+    targets (e.g. two relationships would both write to the same column) for
+    sibling classes in an inheritance hierarchy, where the two relationships
+    would never actually conflict during writes.

lib/sqlalchemy/orm/relationships.py

@@ -1799,6 +1799,15 @@ class RelationshipProperty(StrategizedProperty):
                 (self.key, self.parent.class_)
             )
 
+    def _persists_for(self, mapper):
+        """Return True if this property will persist values on behalf
+        of the given mapper.
+
+        """
+
+        return self.key in mapper.relationships and \
+            mapper.relationships[self.key] is self
+
     def _columns_are_mapped(self, *cols):
         """Return True if all columns in the given collection are
         mapped by the tables referenced by this :class:`.Relationship`.
@@ -2673,10 +2682,16 @@ class JoinCondition(object):
             else:
                 other_props = []
                 prop_to_from = self._track_overlapping_sync_targets[to_]
+
                 for pr, fr_ in prop_to_from.items():
                     if pr.mapper in mapperlib._mapper_registry and \
+                        (
+                            self.prop._persists_for(pr.parent) or
+                            pr._persists_for(self.prop.parent)
+                        ) and \
                         fr_ is not from_ and \
                             pr not in self.prop._reverse_property:
+
                         other_props.append((pr, fr_))
 
                 if other_props:

test/orm/test_relationships.py

@@ -16,6 +16,8 @@ from sqlalchemy.testing import fixtures
 from test.orm import _fixtures
 from sqlalchemy import exc
 from sqlalchemy import inspect
+from sqlalchemy import ForeignKeyConstraint
+from sqlalchemy.ext.declarative import declarative_base
 
 
 class _RelationshipErrors(object):
@@ -528,6 +530,166 @@ class DirectSelfRefFKTest(fixtures.MappedTest, AssertsCompiledSQL):
         )
 
 
+class OverlappingFksSiblingTest(fixtures.TestBase):
+    """Test multiple relationships that use sections of the same
+    composite foreign key.
+
+    """
+
+    def teardown(self):
+        clear_mappers()
+
+    def _fixture_one(
+            self, add_b_a=False, add_b_a_viewonly=False, add_b_amember=False,
+            add_bsub1_a=False, add_bsub2_a_viewonly=False):
+
+        Base = declarative_base(metadata=self.metadata)
+
+        class A(Base):
+
+            __tablename__ = 'a'
+
+            id = Column(Integer, primary_key=True)
+            a_members = relationship('AMember', backref='a')
+
+        class AMember(Base):
+
+            __tablename__ = 'a_member'
+
+            a_id = Column(Integer, ForeignKey('a.id'), primary_key=True)
+            a_member_id = Column(Integer, primary_key=True)
+
+        class B(Base):
+            __tablename__ = 'b'
+
+            __mapper_args__ = {
+                'polymorphic_on': 'type'
+            }
+
+            id = Column(Integer, primary_key=True)
+            type = Column(String(20))
+
+            a_id = Column(Integer, ForeignKey('a.id'), nullable=False)
+            a_member_id = Column(Integer)
+
+            __table_args__ = (
+                ForeignKeyConstraint(
+                    ('a_id', 'a_member_id'),
+                    ('a_member.a_id', 'a_member.a_member_id')),
+            )
+
+            # if added and viewonly is not true, this relationship
+            # writes to B.a_id, which conflicts with BSub2.a_member,
+            # so should warn
+            if add_b_a:
+                a = relationship('A', viewonly=add_b_a_viewonly)
+
+            # if added, this relationship writes to B.a_id, which conflicts
+            # with BSub1.a
+            if add_b_amember:
+                a_member = relationship('AMember')
+
+        # however, *no* warning should be emitted otherwise.
+
+        class BSub1(B):
+
+            if add_bsub1_a:
+                a = relationship('A')
+
+            __mapper_args__ = {'polymorphic_identity': 'bsub1'}
+
+        class BSub2(B):
+
+            if add_bsub2_a_viewonly:
+                a = relationship("A", viewonly=True)
+
+            a_member = relationship('AMember')
+
+            __mapper_args__ = {'polymorphic_identity': 'bsub2'}
+
+        configure_mappers()
+        self.metadata.create_all()
+
+        return A, AMember, B, BSub1, BSub2
+
+    @testing.provide_metadata
+    def _test_fixture_one_run(self, **kw):
+        A, AMember, B, BSub1, BSub2 = self._fixture_one(**kw)
+
+        bsub2 = BSub2()
+        am1 = AMember(a_member_id=1)
+
+        a1 = A(a_members=[am1])
+        bsub2.a_member = am1
+
+        bsub1 = BSub1()
+        a2 = A()
+        bsub1.a = a2
+
+        session = Session(testing.db)
+        session.add_all([bsub1, bsub2, am1, a1, a2])
+        session.commit()
+
+        assert bsub1.a is a2
+        assert bsub2.a is a1
+
+        # meaningless, because BSub1 doesn't have a_member
+        bsub1.a_member = am1
+
+        # meaningless, because BSub2's ".a" is viewonly=True
+        bsub2.a = a2
+
+        session.commit()
+        assert bsub1.a is a2  # beacuse bsub1.a_member is not a relationship
+        assert bsub2.a is a1  # because bsub2.a is viewonly=True
+
+        # everyone has a B.a relationship
+        eq_(
+            session.query(B, A).outerjoin(B.a).order_by(B.id).all(),
+            [(bsub1, a2), (bsub2, a1)]
+        )
+
+    @testing.provide_metadata
+    def test_warn_one(self):
+        assert_raises_message(
+            exc.SAWarning,
+            r"relationship '(?:BSub1.a|BSub2.a_member|B.a)' will copy column "
+            r"(?:a.id|a_member.a_id) to column b.a_id",
+            self._fixture_one, add_b_a=True, add_bsub1_a=True
+        )
+
+    @testing.provide_metadata
+    def test_warn_two(self):
+        assert_raises_message(
+            exc.SAWarning,
+            r"relationship '(?:BSub1.a|B.a_member)' will copy column "
+            r"(?:a.id|a_member.a_id) to column b.a_id",
+            self._fixture_one, add_b_amember=True, add_bsub1_a=True
+        )
+
+    @testing.provide_metadata
+    def test_warn_three(self):
+        assert_raises_message(
+            exc.SAWarning,
+            r"relationship '(?:BSub1.a|B.a_member|B.a)' will copy column "
+            r"(?:a.id|a_member.a_id) to column b.a_id",
+            self._fixture_one, add_b_amember=True, add_bsub1_a=True,
+            add_b_a=True
+        )
+
+    @testing.provide_metadata
+    def test_works_one(self):
+        self._test_fixture_one_run(
+            add_b_a=True, add_b_a_viewonly=True, add_bsub1_a=True
+        )
+
+    @testing.provide_metadata
+    def test_works_two(self):
+        self._test_fixture_one_run(
+            add_b_a=True, add_bsub2_a_viewonly=True
+        )
+
+
 class CompositeSelfRefFKTest(fixtures.MappedTest, AssertsCompiledSQL):
 
     """Tests a composite FK where, in