- It is now an error to specify both columns of a binary primaryjoin

condition in the foreign_keys or remote_side collection.  Whereas
previously it was just nonsensical, but would succeed in a
non-deterministic way.

CHANGES

@@ -66,6 +66,11 @@ CHANGES
     - Fixed Query.update() and Query.delete() failures with eagerloaded
       relations. [ticket:1378]
 
+    - It is now an error to specify both columns of a binary primaryjoin
+      condition in the foreign_keys or remote_side collection.  Whereas
+      previously it was just nonsensical, but would succeed in a 
+      non-deterministic way.
+      
 - schema
     - Added a quote_schema() method to the IdentifierPreparer class
       so that dialects can override how schemas get handled. This

lib/sqlalchemy/orm/properties.py

@@ -839,8 +839,8 @@ class RelationProperty(StrategizedProperty):
                     if self._foreign_keys:
                         raise sa_exc.ArgumentError("Could not determine relation direction for "
                             "primaryjoin condition '%s', on relation %s. "
-                            "Are the columns in 'foreign_keys' present within the given "
-                            "join condition ?" % (self.primaryjoin, self))
+                            "Do the columns in 'foreign_keys' represent only the 'foreign' columns "
+                            "in this join condition ?" % (self.primaryjoin, self))
                     else:
                         raise sa_exc.ArgumentError("Could not determine relation direction for "
                             "primaryjoin condition '%s', on relation %s. "

lib/sqlalchemy/sql/util.py

@@ -343,14 +343,14 @@ def criterion_as_pairs(expression, consider_as_foreign_keys=None, consider_as_re
             return
 
         if consider_as_foreign_keys:
-            if binary.left in consider_as_foreign_keys:
+            if binary.left in consider_as_foreign_keys and (binary.right is binary.left or binary.right not in consider_as_foreign_keys):
                 pairs.append((binary.right, binary.left))
-            elif binary.right in consider_as_foreign_keys:
+            elif binary.right in consider_as_foreign_keys and (binary.left is binary.right or binary.left not in consider_as_foreign_keys):
                 pairs.append((binary.left, binary.right))
         elif consider_as_referenced_keys:
-            if binary.left in consider_as_referenced_keys:
+            if binary.left in consider_as_referenced_keys and (binary.right is binary.left or binary.right not in consider_as_referenced_keys):
                 pairs.append((binary.left, binary.right))
-            elif binary.right in consider_as_referenced_keys:
+            elif binary.right in consider_as_referenced_keys and (binary.left is binary.right or binary.left not in consider_as_referenced_keys):
                 pairs.append((binary.right, binary.left))
         else:
             if isinstance(binary.left, schema.Column) and isinstance(binary.right, schema.Column):

test/orm/relationships.py

@@ -835,7 +835,7 @@ class JoinConditionErrorTest(testing.TestBase):
         mapper(C2, t3)
         
         self.assertRaises(sa.exc.NoReferencedColumnError, compile_mappers)
-
+    
     def test_join_error_raised(self):
         m = MetaData()
         t1 = Table('t1', m, 
@@ -1640,6 +1640,53 @@ class InvalidRelationEscalationTest(_base.MappedTest):
             "Could not locate any equated, locally mapped column pairs "
             "for primaryjoin condition", sa.orm.compile_mappers)
 
+    @testing.resolve_artifact_names
+    def test_ambiguous_fks(self):
+        mapper(Foo, foos, properties={
+            'bars':relation(Bar,
+                            primaryjoin=foos.c.id==bars.c.fid,
+                            foreign_keys=[foos.c.id, bars.c.fid])})
+        mapper(Bar, bars)
+
+        self.assertRaisesMessage(
+            sa.exc.ArgumentError, 
+                "Do the columns in 'foreign_keys' represent only the "
+                "'foreign' columns in this join condition ?", 
+                sa.orm.compile_mappers)
+
+    @testing.resolve_artifact_names
+    def test_ambiguous_remoteside_o2m(self):
+        mapper(Foo, foos, properties={
+            'bars':relation(Bar,
+                            primaryjoin=foos.c.id==bars.c.fid,
+                            foreign_keys=[bars.c.fid],
+                            remote_side=[foos.c.id, bars.c.fid],
+                            viewonly=True
+                            )})
+        mapper(Bar, bars)
+
+        self.assertRaisesMessage(
+            sa.exc.ArgumentError, 
+                "could not determine any local/remote column pairs",
+                sa.orm.compile_mappers)
+
+    @testing.resolve_artifact_names
+    def test_ambiguous_remoteside_m2o(self):
+        mapper(Foo, foos, properties={
+            'bars':relation(Bar,
+                            primaryjoin=foos.c.id==bars.c.fid,
+                            foreign_keys=[foos.c.id],
+                            remote_side=[foos.c.id, bars.c.fid],
+                            viewonly=True
+                            )})
+        mapper(Bar, bars)
+
+        self.assertRaisesMessage(
+            sa.exc.ArgumentError, 
+                "could not determine any local/remote column pairs",
+                sa.orm.compile_mappers)
+        
+    
     @testing.resolve_artifact_names
     def test_no_equated_self_ref(self):
         mapper(Foo, foos, properties={