- merged #1401 branch from bitbucket

- resolved some serious speed hits I missed, we need to ensure
only deannotated columns are used in the local/remote collections and soforth
so that hash lookups against mapped columns don't dig into __eq__()
- fix some other parity mismatches regarding stuff from [ticket:2453],
including finding another case where _deep_annotate() was doing the wrong thing,
new tests.
- [feature] Major rewrite of relationship()
internals now allow join conditions which
include columns pointing to themselves
within composite foreign keys.   A new
API for very specialized primaryjoin conditions
is added, allowing conditions based on
SQL functions, CAST, etc. to be handled
by placing the annotation functions
remote() and foreign() inline within the
expression when necessary.  Previous recipes
using the semi-private _local_remote_pairs
approach can be upgraded to this new
approach. [ticket:1401]

11 files changed, 1682 insertions, 250 deletions

diff --git a/test/aaa_profiling/test_orm.py b/test/aaa_profiling/test_orm.py
index 8cc2e0087..8dea18359 100644
--- a/test/aaa_profiling/test_orm.py
+++ b/test/aaa_profiling/test_orm.py
@@ -95,6 +95,7 @@ class MergeTest(fixtures.MappedTest):
                                         '3':1050}
                             )
         def go():
+            print "GO"
             p2 = sess2.merge(p1)
         go()
 
diff --git a/test/ext/test_declarative.py b/test/ext/test_declarative.py
index 69042b5c8..5e185f664 100644
--- a/test/ext/test_declarative.py
+++ b/test/ext/test_declarative.py
@@ -368,6 +368,28 @@ class DeclarativeTest(DeclarativeTestBase):
         assert class_mapper(User).get_property('props').secondary \
             is user_to_prop
 
+    def test_string_dependency_resolution_annotations(self):
+        Base = decl.declarative_base()
+
+        class Parent(Base):
+            __tablename__ = 'parent'
+            id = Column(Integer, primary_key=True)
+            name = Column(String)
+            children = relationship("Child",
+                    primaryjoin="Parent.name==remote(foreign(func.lower(Child.name_upper)))"
+                )
+
+        class Child(Base):
+            __tablename__ = 'child'
+            id = Column(Integer, primary_key=True)
+            name_upper = Column(String)
+
+        configure_mappers()
+        eq_(
+            Parent.children.property._calculated_foreign_keys,
+            set([Child.name_upper.property.columns[0]])
+        )
+
     def test_shared_class_registry(self):
         reg = {}
         Base1 = decl.declarative_base(testing.db, class_registry=reg)
diff --git a/test/ext/test_serializer.py b/test/ext/test_serializer.py
index 5134d71ee..87b7a2f67 100644
--- a/test/ext/test_serializer.py
+++ b/test/ext/test_serializer.py
@@ -112,13 +112,18 @@ class SerializeTest(fixtures.MappedTest):
         eq_(q2.all(), [Address(email='ed@wood.com'),
             Address(email='ed@lala.com'),
             Address(email='ed@bettyboop.com')])
-        q = \
-            Session.query(User).join(User.addresses).\
-                filter(Address.email.like('%fred%'))
-        q2 = serializer.loads(serializer.dumps(q, -1), users.metadata,
-                              Session)
-        eq_(q2.all(), [User(name='fred')])
-        eq_(list(q2.values(User.id, User.name)), [(9, u'fred')])
+
+        # unfortunately pickle just doesn't have the horsepower
+        # to pickle annotated joins, both cpickle and pickle
+        # get confused likely since identity-unequal/hash equal
+        # objects with cycles being used
+        #q = \
+        #    Session.query(User).join(User.addresses).\
+        #       filter(Address.email.like('%fred%'))
+        #q2 = serializer.loads(serializer.dumps(q, -1), users.metadata,
+        #                      Session)
+        #eq_(q2.all(), [User(name='fred')])
+        #eq_(list(q2.values(User.id, User.name)), [(9, u'fred')])
 
     @testing.exclude('sqlite', '<=', (3, 5, 9),
                      'id comparison failing on the buildbot')
diff --git a/test/orm/inheritance/test_abc_inheritance.py b/test/orm/inheritance/test_abc_inheritance.py
index 6a2f579ae..e1304e26e 100644
--- a/test/orm/inheritance/test_abc_inheritance.py
+++ b/test/orm/inheritance/test_abc_inheritance.py
@@ -111,7 +111,11 @@ def produce_test(parent, child, direction):
             parent_class = parent_mapper.class_
             child_class = child_mapper.class_
 
-            parent_mapper.add_property("collection", relationship(child_mapper, primaryjoin=relationshipjoin, foreign_keys=foreign_keys, remote_side=remote_side, uselist=True))
+            parent_mapper.add_property("collection", 
+                                relationship(child_mapper, 
+                                            primaryjoin=relationshipjoin, 
+                                            foreign_keys=foreign_keys, 
+                                            remote_side=remote_side, uselist=True))
 
             sess = create_session()
 
diff --git a/test/orm/test_joins.py b/test/orm/test_joins.py
index db7c78cdd..6c43a2f39 100644
--- a/test/orm/test_joins.py
+++ b/test/orm/test_joins.py
@@ -1700,21 +1700,29 @@ class SelfReferentialTest(fixtures.MappedTest, AssertsCompiledSQL):
         sess.flush()
         sess.close()
 
-    def test_join(self):
+    def test_join_1(self):
         Node = self.classes.Node
-
         sess = create_session()
 
         node = sess.query(Node).join('children', aliased=True).filter_by(data='n122').first()
         assert node.data=='n12'
 
+    def test_join_2(self):
+        Node = self.classes.Node
+        sess = create_session()
         ret = sess.query(Node.data).join(Node.children, aliased=True).filter_by(data='n122').all()
         assert ret == [('n12',)]
 
 
+    def test_join_3(self):
+        Node = self.classes.Node
+        sess = create_session()
         node = sess.query(Node).join('children', 'children', aliased=True).filter_by(data='n122').first()
         assert node.data=='n1'
 
+    def test_join_4(self):
+        Node = self.classes.Node
+        sess = create_session()
         node = sess.query(Node).filter_by(data='n122').join('parent', aliased=True).filter_by(data='n12').\
             join('parent', aliased=True, from_joinpoint=True).filter_by(data='n1').first()
         assert node.data == 'n122'
diff --git a/test/orm/test_mapper.py b/test/orm/test_mapper.py
index 79ae7ff59..4478e5d80 100644
--- a/test/orm/test_mapper.py
+++ b/test/orm/test_mapper.py
@@ -522,6 +522,8 @@ class MapperTest(_fixtures.FixtureTest, AssertsCompiledSQL):
 
         assert User.x.property.columns[0] is not expr
         assert User.x.property.columns[0].element.left is users.c.name
+        # a deannotate needs to clone the base, in case 
+        # the original one referenced annotated elements.
         assert User.x.property.columns[0].element.right is not expr.right
 
         assert User.y.property.columns[0] is not expr2
diff --git a/test/orm/test_query.py b/test/orm/test_query.py
index bcc976816..1b57299f0 100644
--- a/test/orm/test_query.py
+++ b/test/orm/test_query.py
@@ -623,6 +623,14 @@ class OperatorTest(QueryTest, AssertsCompiledSQL):
 
         self._test(Address.user != None, "addresses.user_id IS NOT NULL")
 
+    def test_foo(self):
+        Node = self.classes.Node
+        nalias = aliased(Node)
+        self._test(
+            nalias.parent.has(Node.data=='some data'), 
+           "EXISTS (SELECT 1 FROM nodes WHERE nodes.id = nodes_1.parent_id AND nodes.data = :data_1)"
+        )
+
     def test_selfref_relationship(self):
         Node = self.classes.Node
 
diff --git a/test/orm/test_rel_fn.py b/test/orm/test_rel_fn.py
new file mode 100644
index 000000000..e35dd925a
--- /dev/null
+++ b/test/orm/test_rel_fn.py
@@ -0,0 +1,927 @@
+from test.lib.testing import assert_raises, assert_raises_message, eq_, \
+    AssertsCompiledSQL, is_
+from test.lib import fixtures
+from sqlalchemy.orm import relationships, foreign, remote, remote_foreign
+from sqlalchemy import MetaData, Table, Column, ForeignKey, Integer, \
+    select, ForeignKeyConstraint, exc, func, and_
+from sqlalchemy.orm.interfaces import ONETOMANY, MANYTOONE, MANYTOMANY
+
+
+class _JoinFixtures(object):
+    @classmethod
+    def setup_class(cls):
+        m = MetaData()
+        cls.left = Table('lft', m,
+            Column('id', Integer, primary_key=True),
+            Column('x', Integer),
+            Column('y', Integer),
+        )
+        cls.right = Table('rgt', m,
+            Column('id', Integer, primary_key=True),
+            Column('lid', Integer, ForeignKey('lft.id')),
+            Column('x', Integer),
+            Column('y', Integer),
+        )
+        cls.right_multi_fk = Table('rgt_multi_fk', m,
+            Column('id', Integer, primary_key=True),
+            Column('lid1', Integer, ForeignKey('lft.id')),
+            Column('lid2', Integer, ForeignKey('lft.id')),
+        )
+
+        cls.selfref = Table('selfref', m,
+            Column('id', Integer, primary_key=True),
+            Column('sid', Integer, ForeignKey('selfref.id'))
+        )
+        cls.composite_selfref = Table('composite_selfref', m,
+            Column('id', Integer, primary_key=True),
+            Column('group_id', Integer, primary_key=True),
+            Column('parent_id', Integer),
+            ForeignKeyConstraint(
+                ['parent_id', 'group_id'],
+                ['composite_selfref.id', 'composite_selfref.group_id']
+            )
+        )
+        cls.m2mleft = Table('m2mlft', m,
+            Column('id', Integer, primary_key=True),
+        )
+        cls.m2mright = Table('m2mrgt', m,
+            Column('id', Integer, primary_key=True),
+        )
+        cls.m2msecondary = Table('m2msecondary', m,
+            Column('lid', Integer, ForeignKey('m2mlft.id'), primary_key=True),
+            Column('rid', Integer, ForeignKey('m2mrgt.id'), primary_key=True),
+        )
+        cls.m2msecondary_no_fks = Table('m2msecondary_no_fks', m,
+            Column('lid', Integer, primary_key=True),
+            Column('rid', Integer, primary_key=True),
+        )
+        cls.m2msecondary_ambig_fks = Table('m2msecondary_ambig_fks', m,
+            Column('lid1', Integer, ForeignKey('m2mlft.id'), primary_key=True),
+            Column('rid1', Integer, ForeignKey('m2mrgt.id'), primary_key=True),
+            Column('lid2', Integer, ForeignKey('m2mlft.id'), primary_key=True),
+            Column('rid2', Integer, ForeignKey('m2mrgt.id'), primary_key=True),
+        )
+        cls.base_w_sub_rel = Table('base_w_sub_rel', m,
+            Column('id', Integer, primary_key=True),
+            Column('sub_id', Integer, ForeignKey('rel_sub.id'))
+        )
+        cls.rel_sub = Table('rel_sub', m,
+            Column('id', Integer, ForeignKey('base_w_sub_rel.id'), 
+                                primary_key=True)
+        )
+        cls.base = Table('base', m,
+            Column('id', Integer, primary_key=True),
+        )
+        cls.sub = Table('sub', m,
+            Column('id', Integer, ForeignKey('base.id'), 
+                                primary_key=True),
+        )
+        cls.sub_w_base_rel = Table('sub_w_base_rel', m,
+            Column('id', Integer, ForeignKey('base.id'), 
+                                primary_key=True),
+            Column('base_id', Integer, ForeignKey('base.id'))
+        )
+        cls.right_w_base_rel = Table('right_w_base_rel', m,
+            Column('id', Integer, primary_key=True),
+            Column('base_id', Integer, ForeignKey('base.id'))
+        )
+
+        cls.three_tab_a = Table('three_tab_a', m, 
+            Column('id', Integer, primary_key=True),
+        )
+        cls.three_tab_b = Table('three_tab_b', m, 
+            Column('id', Integer, primary_key=True),
+            Column('aid', Integer, ForeignKey('three_tab_a.id'))
+        )
+        cls.three_tab_c = Table('three_tab_c', m, 
+            Column('id', Integer, primary_key=True),
+            Column('aid', Integer, ForeignKey('three_tab_a.id')),
+            Column('bid', Integer, ForeignKey('three_tab_b.id'))
+        )
+
+    def _join_fixture_overlapping_three_tables(self, **kw):
+        def _can_sync(*cols):
+            for c in cols:
+                if self.three_tab_c.c.contains_column(c):
+                    return False
+            else:
+                return True
+        return relationships.JoinCondition(
+            self.three_tab_a, 
+            self.three_tab_b,
+            self.three_tab_a, 
+            self.three_tab_b,
+            support_sync=False,
+            can_be_synced_fn=_can_sync,
+            primaryjoin=and_(
+                self.three_tab_a.c.id==self.three_tab_b.c.aid,
+                self.three_tab_c.c.bid==self.three_tab_b.c.id,
+                self.three_tab_c.c.aid==self.three_tab_a.c.id
+            )
+        )
+
+    def _join_fixture_m2m(self, **kw):
+        return relationships.JoinCondition(
+                    self.m2mleft, 
+                    self.m2mright, 
+                    self.m2mleft, 
+                    self.m2mright,
+                    secondary=self.m2msecondary,
+                    **kw
+                )
+
+    def _join_fixture_o2m(self, **kw):
+        return relationships.JoinCondition(
+                    self.left, 
+                    self.right, 
+                    self.left, 
+                    self.right,
+                    **kw
+                )
+
+    def _join_fixture_m2o(self, **kw):
+        return relationships.JoinCondition(
+                    self.right, 
+                    self.left, 
+                    self.right,
+                    self.left,
+                    **kw
+                )
+
+    def _join_fixture_o2m_selfref(self, **kw):
+        return relationships.JoinCondition(
+            self.selfref,
+            self.selfref,
+            self.selfref,
+            self.selfref,
+            **kw
+        )
+
+    def _join_fixture_m2o_selfref(self, **kw):
+        return relationships.JoinCondition(
+            self.selfref,
+            self.selfref,
+            self.selfref,
+            self.selfref,
+            remote_side=set([self.selfref.c.id]),
+            **kw
+        )
+
+    def _join_fixture_o2m_composite_selfref(self, **kw):
+        return relationships.JoinCondition(
+            self.composite_selfref,
+            self.composite_selfref,
+            self.composite_selfref,
+            self.composite_selfref,
+            **kw
+        )
+
+    def _join_fixture_m2o_composite_selfref(self, **kw):
+        return relationships.JoinCondition(
+            self.composite_selfref,
+            self.composite_selfref,
+            self.composite_selfref,
+            self.composite_selfref,
+            remote_side=set([self.composite_selfref.c.id, 
+                            self.composite_selfref.c.group_id]),
+            **kw
+        )
+
+    def _join_fixture_o2m_composite_selfref_func(self, **kw):
+        return relationships.JoinCondition(
+            self.composite_selfref,
+            self.composite_selfref,
+            self.composite_selfref,
+            self.composite_selfref,
+            primaryjoin=and_(
+                self.composite_selfref.c.group_id==
+                    func.foo(self.composite_selfref.c.group_id),
+                self.composite_selfref.c.parent_id==
+                    self.composite_selfref.c.id
+            ),
+            **kw
+        )
+
+    def _join_fixture_o2m_composite_selfref_func_annotated(self, **kw):
+        return relationships.JoinCondition(
+            self.composite_selfref,
+            self.composite_selfref,
+            self.composite_selfref,
+            self.composite_selfref,
+            primaryjoin=and_(
+                remote(self.composite_selfref.c.group_id)==
+                    func.foo(self.composite_selfref.c.group_id),
+                remote(self.composite_selfref.c.parent_id)==
+                    self.composite_selfref.c.id
+            ),
+            **kw
+        )
+
+    def _join_fixture_compound_expression_1(self, **kw):
+        return relationships.JoinCondition(
+            self.left,
+            self.right,
+            self.left,
+            self.right,
+            primaryjoin=(self.left.c.x + self.left.c.y) == \
+                            relationships.remote_foreign(
+                                self.right.c.x * self.right.c.y
+                            ),
+            **kw
+        )
+
+    def _join_fixture_compound_expression_2(self, **kw):
+        return relationships.JoinCondition(
+            self.left,
+            self.right,
+            self.left,
+            self.right,
+            primaryjoin=(self.left.c.x + self.left.c.y) == \
+                            relationships.foreign(
+                                self.right.c.x * self.right.c.y
+                            ),
+            **kw
+        )
+
+    def _join_fixture_compound_expression_1_non_annotated(self, **kw):
+        return relationships.JoinCondition(
+            self.left,
+            self.right,
+            self.left,
+            self.right,
+            primaryjoin=(self.left.c.x + self.left.c.y) == \
+                            (
+                                self.right.c.x * self.right.c.y
+                            ),
+            **kw
+        )
+
+    def _join_fixture_base_to_joined_sub(self, **kw):
+        # see test/orm/inheritance/test_abc_inheritance:TestaTobM2O
+        # and others there
+        right = self.base_w_sub_rel.join(self.rel_sub,
+            self.base_w_sub_rel.c.id==self.rel_sub.c.id
+        )
+        return relationships.JoinCondition(
+            self.base_w_sub_rel,
+            right,
+            self.base_w_sub_rel,
+            self.rel_sub,
+            primaryjoin=self.base_w_sub_rel.c.sub_id==\
+                        self.rel_sub.c.id,
+            **kw
+        )
+
+    def _join_fixture_o2m_joined_sub_to_base(self, **kw):
+        left = self.base.join(self.sub_w_base_rel, 
+                        self.base.c.id==self.sub_w_base_rel.c.id)
+        return relationships.JoinCondition(
+            left,
+            self.base,
+            self.sub_w_base_rel,
+            self.base,
+            primaryjoin=self.sub_w_base_rel.c.base_id==self.base.c.id
+        )
+
+    def _join_fixture_m2o_sub_to_joined_sub(self, **kw):
+        # see test.orm.test_mapper:MapperTest.test_add_column_prop_deannotate,
+        right = self.base.join(self.right_w_base_rel, 
+                        self.base.c.id==self.right_w_base_rel.c.id)
+        return relationships.JoinCondition(
+            self.right_w_base_rel,
+            right,
+            self.right_w_base_rel,
+            self.right_w_base_rel,
+        )
+
+    def _join_fixture_m2o_sub_to_joined_sub_func(self, **kw):
+        # see test.orm.test_mapper:MapperTest.test_add_column_prop_deannotate,
+        right = self.base.join(self.right_w_base_rel, 
+                        self.base.c.id==self.right_w_base_rel.c.id)
+        return relationships.JoinCondition(
+            self.right_w_base_rel,
+            right,
+            self.right_w_base_rel,
+            self.right_w_base_rel,
+            primaryjoin=self.right_w_base_rel.c.base_id==\
+                func.foo(self.base.c.id)
+        )
+
+    def _join_fixture_o2o_joined_sub_to_base(self, **kw):
+        left = self.base.join(self.sub, 
+                        self.base.c.id==self.sub.c.id)
+
+        # see test_relationships->AmbiguousJoinInterpretedAsSelfRef
+        return relationships.JoinCondition(
+            left,
+            self.sub,
+            left, 
+            self.sub,
+        )
+
+    def _join_fixture_o2m_to_annotated_func(self, **kw):
+        return relationships.JoinCondition(
+                    self.left, 
+                    self.right, 
+                    self.left, 
+                    self.right,
+                    primaryjoin=self.left.c.id==
+                        foreign(func.foo(self.right.c.lid)),
+                    **kw
+                )
+
+    def _join_fixture_o2m_to_oldstyle_func(self, **kw):
+        return relationships.JoinCondition(
+                    self.left, 
+                    self.right, 
+                    self.left, 
+                    self.right,
+                    primaryjoin=self.left.c.id==
+                        func.foo(self.right.c.lid),
+                    consider_as_foreign_keys=[self.right.c.lid],
+                    **kw
+                )
+
+    def _assert_non_simple_warning(self, fn):
+        assert_raises_message(
+            exc.SAWarning,
+            "Non-simple column elements in "
+            "primary join condition for property "
+            r"None - consider using remote\(\) "
+            "annotations to mark the remote side.",
+            fn
+        )
+
+    def _assert_raises_no_relevant_fks(self, fn, expr, relname, 
+        primary, *arg, **kw):
+        assert_raises_message(
+            exc.ArgumentError, 
+            r"Could not locate any relevant foreign key columns "
+            r"for %s join condition '%s' on relationship %s.  "
+            r"Ensure that referencing columns are associated with "
+            r"a ForeignKey or ForeignKeyConstraint, or are annotated "
+            r"in the join condition with the foreign\(\) annotation."
+            % (
+                primary, expr, relname
+            ),
+            fn, *arg, **kw
+        )
+
+    def _assert_raises_no_equality(self, fn, expr, relname, 
+        primary, *arg, **kw):
+        assert_raises_message(
+            sa.exc.ArgumentError, 
+            "Could not locate any simple equality expressions "
+            "involving locally mapped foreign key columns for %s join "
+            "condition '%s' on relationship %s.  "
+            "Ensure that referencing columns are associated with a "
+            "ForeignKey or ForeignKeyConstraint, or are annotated in "
+            r"the join condition with the foreign\(\) annotation. "
+            "To allow comparison operators other than '==', "
+            "the relationship can be marked as viewonly=True." % (
+                primary, expr, relname
+            ),
+            fn, *arg, **kw
+        )
+
+    def _assert_raises_ambig_join(self, fn, relname, secondary_arg,
+        *arg, **kw):
+        if secondary_arg is not None:
+            assert_raises_message(
+                exc.AmbiguousForeignKeysError,
+                "Could not determine join condition between "
+                "parent/child tables on relationship %s - "
+                "there are multiple foreign key paths linking the "
+                "tables via secondary table '%s'.  "
+                "Specify the 'foreign_keys' argument, providing a list "
+                "of those columns which should be counted as "
+                "containing a foreign key reference from the "
+                "secondary table to each of the parent and child tables."
+                % (relname, secondary_arg),
+                fn, *arg, **kw)
+        else:
+            assert_raises_message(
+                exc.AmbiguousForeignKeysError,
+                "Could not determine join condition between "
+                "parent/child tables on relationship %s - "
+                "there are no foreign keys linking these tables.  " 
+                % (relname,),
+                fn, *arg, **kw)
+
+    def _assert_raises_no_join(self, fn, relname, secondary_arg,
+        *arg, **kw):
+        if secondary_arg is not None:
+            assert_raises_message(
+                exc.NoForeignKeysError,
+                "Could not determine join condition between "
+                "parent/child tables on relationship %s - "
+                "there are no foreign keys linking these tables "
+                "via secondary table '%s'.  "
+                "Ensure that referencing columns are associated with a ForeignKey "
+                "or ForeignKeyConstraint, or specify 'primaryjoin' and "
+                "'secondaryjoin' expressions"
+                % (relname, secondary_arg),
+                fn, *arg, **kw)
+        else:
+            assert_raises_message(
+                exc.NoForeignKeysError,
+                "Could not determine join condition between "
+                "parent/child tables on relationship %s - "
+                "there are no foreign keys linking these tables.  "
+                "Ensure that referencing columns are associated with a ForeignKey "
+                "or ForeignKeyConstraint, or specify a 'primaryjoin' "
+                "expression."
+                % (relname,),
+                fn, *arg, **kw)
+
+
+class ColumnCollectionsTest(_JoinFixtures, fixtures.TestBase, AssertsCompiledSQL):
+    def test_determine_local_remote_pairs_o2o_joined_sub_to_base(self):
+        joincond = self._join_fixture_o2o_joined_sub_to_base()
+        eq_(
+            joincond.local_remote_pairs,
+            [(self.base.c.id, self.sub.c.id)]
+        )
+
+    def test_determine_synchronize_pairs_o2m_to_annotated_func(self):
+        joincond = self._join_fixture_o2m_to_annotated_func()
+        eq_(
+            joincond.synchronize_pairs,
+            [(self.left.c.id, self.right.c.lid)]
+        )
+
+    def test_determine_synchronize_pairs_o2m_to_oldstyle_func(self):
+        joincond = self._join_fixture_o2m_to_oldstyle_func()
+        eq_(
+            joincond.synchronize_pairs,
+            [(self.left.c.id, self.right.c.lid)]
+        )
+
+    def test_determine_local_remote_base_to_joined_sub(self):
+        joincond = self._join_fixture_base_to_joined_sub()
+        eq_(
+            joincond.local_remote_pairs,
+            [
+                (self.base_w_sub_rel.c.sub_id, self.rel_sub.c.id)
+            ]
+        )
+
+    def test_determine_local_remote_o2m_joined_sub_to_base(self):
+        joincond = self._join_fixture_o2m_joined_sub_to_base()
+        eq_(
+            joincond.local_remote_pairs,
+            [
+                (self.sub_w_base_rel.c.base_id, self.base.c.id)
+            ]
+        )
+
+    def test_determine_local_remote_m2o_sub_to_joined_sub(self):
+        joincond = self._join_fixture_m2o_sub_to_joined_sub()
+        eq_(
+            joincond.local_remote_pairs,
+            [
+                (self.right_w_base_rel.c.base_id, self.base.c.id)
+            ]
+        )
+
+    def test_determine_remote_columns_compound_1(self):
+        joincond = self._join_fixture_compound_expression_1(
+                                support_sync=False)
+        eq_(
+            joincond.remote_columns,
+            set([self.right.c.x, self.right.c.y])
+        )
+
+    def test_determine_local_remote_compound_1(self):
+        joincond = self._join_fixture_compound_expression_1(
+                                support_sync=False)
+        eq_(
+            joincond.local_remote_pairs,
+            [
+                (self.left.c.x, self.right.c.x), 
+                (self.left.c.x, self.right.c.y), 
+                (self.left.c.y, self.right.c.x),
+                (self.left.c.y, self.right.c.y)
+            ]
+        )
+
+    def test_determine_local_remote_compound_2(self):
+        joincond = self._join_fixture_compound_expression_2(
+                                support_sync=False)
+        eq_(
+            joincond.local_remote_pairs,
+            [
+                (self.left.c.x, self.right.c.x), 
+                (self.left.c.x, self.right.c.y), 
+                (self.left.c.y, self.right.c.x),
+                (self.left.c.y, self.right.c.y)
+            ]
+        )
+
+    def test_determine_local_remote_compound_1(self):
+        joincond = self._join_fixture_compound_expression_1()
+        eq_(
+            joincond.local_remote_pairs,
+            [
+                (self.left.c.x, self.right.c.x),
+                (self.left.c.x, self.right.c.y),
+                (self.left.c.y, self.right.c.x),
+                (self.left.c.y, self.right.c.y),
+            ]
+        )
+
+    def test_err_local_remote_compound_1(self):
+        self._assert_raises_no_relevant_fks(
+            self._join_fixture_compound_expression_1_non_annotated,
+            r'lft.x \+ lft.y = rgt.x \* rgt.y',
+            "None", "primary"
+        )
+
+    def test_determine_remote_columns_compound_2(self):
+        joincond = self._join_fixture_compound_expression_2(
+                                support_sync=False)
+        eq_(
+            joincond.remote_columns,
+            set([self.right.c.x, self.right.c.y])
+        )
+
+
+    def test_determine_remote_columns_o2m(self):
+        joincond = self._join_fixture_o2m()
+        eq_(
+            joincond.remote_columns,
+            set([self.right.c.lid])
+        )
+
+    def test_determine_remote_columns_o2m_selfref(self):
+        joincond = self._join_fixture_o2m_selfref()
+        eq_(
+            joincond.remote_columns,
+            set([self.selfref.c.sid])
+        )
+
+    def test_determine_local_remote_pairs_o2m_composite_selfref(self):
+        joincond = self._join_fixture_o2m_composite_selfref()
+        eq_(
+            joincond.local_remote_pairs,
+            [
+                (self.composite_selfref.c.group_id, self.composite_selfref.c.group_id),
+                (self.composite_selfref.c.id, self.composite_selfref.c.parent_id),
+            ]
+        )
+
+    def test_determine_local_remote_pairs_o2m_composite_selfref_func_warning(self):
+        self._assert_non_simple_warning(
+            self._join_fixture_o2m_composite_selfref_func
+        )
+
+    def test_determine_local_remote_pairs_o2m_overlap_func_warning(self):
+        self._assert_non_simple_warning(
+            self._join_fixture_m2o_sub_to_joined_sub_func
+        )
+
+    def test_determine_local_remote_pairs_o2m_composite_selfref_func_annotated(self):
+        joincond = self._join_fixture_o2m_composite_selfref_func_annotated()
+        eq_(
+            joincond.local_remote_pairs,
+            [
+                (self.composite_selfref.c.group_id, self.composite_selfref.c.group_id),
+                (self.composite_selfref.c.id, self.composite_selfref.c.parent_id),
+            ]
+        )
+
+    def test_determine_remote_columns_m2o_composite_selfref(self):
+        joincond = self._join_fixture_m2o_composite_selfref()
+        eq_(
+            joincond.remote_columns,
+            set([self.composite_selfref.c.id, 
+                self.composite_selfref.c.group_id])
+        )
+
+    def test_determine_remote_columns_m2o(self):
+        joincond = self._join_fixture_m2o()
+        eq_(
+            joincond.remote_columns,
+            set([self.left.c.id])
+        )
+
+    def test_determine_local_remote_pairs_o2m(self):
+        joincond = self._join_fixture_o2m()
+        eq_(
+            joincond.local_remote_pairs,
+            [(self.left.c.id, self.right.c.lid)]
+        )
+
+    def test_determine_synchronize_pairs_m2m(self):
+        joincond = self._join_fixture_m2m()
+        eq_(
+            joincond.synchronize_pairs,
+            [(self.m2mleft.c.id, self.m2msecondary.c.lid)]
+        )
+        eq_(
+            joincond.secondary_synchronize_pairs,
+            [(self.m2mright.c.id, self.m2msecondary.c.rid)]
+        )
+
+    def test_determine_local_remote_pairs_o2m_backref(self):
+        joincond = self._join_fixture_o2m()
+        joincond2 = self._join_fixture_m2o(
+            primaryjoin=joincond.primaryjoin_reverse_remote,
+        )
+        eq_(
+            joincond2.local_remote_pairs,
+            [(self.right.c.lid, self.left.c.id)]
+        )
+
+    def test_determine_local_remote_pairs_m2m(self):
+        joincond = self._join_fixture_m2m()
+        eq_(
+            joincond.local_remote_pairs,
+            [(self.m2mleft.c.id, self.m2msecondary.c.lid), 
+            (self.m2mright.c.id, self.m2msecondary.c.rid)]
+        )
+
+    def test_determine_local_remote_pairs_m2m_backref(self):
+        joincond = self._join_fixture_m2m()
+        joincond2 = self._join_fixture_m2m(
+            primaryjoin=joincond.secondaryjoin,
+            secondaryjoin=joincond.primaryjoin
+        )
+        eq_(
+            joincond.local_remote_pairs,
+            [(self.m2mleft.c.id, self.m2msecondary.c.lid), 
+            (self.m2mright.c.id, self.m2msecondary.c.rid)]
+        )
+
+    def test_determine_remote_columns_m2o_selfref(self):
+        joincond = self._join_fixture_m2o_selfref()
+        eq_(
+            joincond.remote_columns,
+            set([self.selfref.c.id])
+        )
+
+    def test_determine_local_remote_cols_three_tab_viewonly(self):
+        joincond = self._join_fixture_overlapping_three_tables()
+        eq_(
+            joincond.local_remote_pairs,
+            [(self.three_tab_a.c.id, self.three_tab_b.c.aid)]
+        )
+        eq_(
+            joincond.remote_columns,
+            set([self.three_tab_b.c.id, self.three_tab_b.c.aid])
+        )
+
+class DirectionTest(_JoinFixtures, fixtures.TestBase, AssertsCompiledSQL):
+    def test_determine_direction_compound_2(self):
+        joincond = self._join_fixture_compound_expression_2(
+                                support_sync=False)
+        is_(
+            joincond.direction,
+            ONETOMANY
+        )
+
+    def test_determine_direction_o2m(self):
+        joincond = self._join_fixture_o2m()
+        is_(joincond.direction, ONETOMANY)
+
+    def test_determine_direction_o2m_selfref(self):
+        joincond = self._join_fixture_o2m_selfref()
+        is_(joincond.direction, ONETOMANY)
+
+    def test_determine_direction_m2o_selfref(self):
+        joincond = self._join_fixture_m2o_selfref()
+        is_(joincond.direction, MANYTOONE)
+
+    def test_determine_direction_o2m_composite_selfref(self):
+        joincond = self._join_fixture_o2m_composite_selfref()
+        is_(joincond.direction, ONETOMANY)
+
+    def test_determine_direction_m2o_composite_selfref(self):
+        joincond = self._join_fixture_m2o_composite_selfref()
+        is_(joincond.direction, MANYTOONE)
+
+    def test_determine_direction_m2o(self):
+        joincond = self._join_fixture_m2o()
+        is_(joincond.direction, MANYTOONE)
+
+
+class DetermineJoinTest(_JoinFixtures, fixtures.TestBase, AssertsCompiledSQL):
+    __dialect__ = 'default'
+
+    def test_determine_join_o2m(self):
+        joincond = self._join_fixture_o2m()
+        self.assert_compile(
+                joincond.primaryjoin,
+                "lft.id = rgt.lid"
+        )
+
+    def test_determine_join_o2m_selfref(self):
+        joincond = self._join_fixture_o2m_selfref()
+        self.assert_compile(
+                joincond.primaryjoin,
+                "selfref.id = selfref.sid"
+        )
+
+    def test_determine_join_m2o_selfref(self):
+        joincond = self._join_fixture_m2o_selfref()
+        self.assert_compile(
+                joincond.primaryjoin,
+                "selfref.id = selfref.sid"
+        )
+
+    def test_determine_join_o2m_composite_selfref(self):
+        joincond = self._join_fixture_o2m_composite_selfref()
+        self.assert_compile(
+                joincond.primaryjoin,
+                "composite_selfref.group_id = composite_selfref.group_id "
+                "AND composite_selfref.id = composite_selfref.parent_id"
+        )
+
+    def test_determine_join_m2o_composite_selfref(self):
+        joincond = self._join_fixture_m2o_composite_selfref()
+        self.assert_compile(
+                joincond.primaryjoin,
+                "composite_selfref.group_id = composite_selfref.group_id "
+                "AND composite_selfref.id = composite_selfref.parent_id"
+        )
+
+    def test_determine_join_m2o(self):
+        joincond = self._join_fixture_m2o()
+        self.assert_compile(
+                joincond.primaryjoin,
+                "lft.id = rgt.lid"
+        )
+
+    def test_determine_join_ambiguous_fks_o2m(self):
+        assert_raises_message(
+            exc.AmbiguousForeignKeysError,
+            "Could not determine join condition between "
+            "parent/child tables on relationship None - "
+            "there are multiple foreign key paths linking "
+            "the tables.  Specify the 'foreign_keys' argument, "
+            "providing a list of those columns which "
+            "should be counted as containing a foreign "
+            "key reference to the parent table.",
+            relationships.JoinCondition,
+                    self.left, 
+                    self.right_multi_fk, 
+                    self.left, 
+                    self.right_multi_fk, 
+        )
+
+    def test_determine_join_no_fks_o2m(self):
+        self._assert_raises_no_join(
+            relationships.JoinCondition,
+            "None", None,
+                    self.left, 
+                    self.selfref, 
+                    self.left, 
+                    self.selfref, 
+        )
+
+
+    def test_determine_join_ambiguous_fks_m2m(self):
+
+        self._assert_raises_ambig_join(
+            relationships.JoinCondition,
+            "None", self.m2msecondary_ambig_fks,
+            self.m2mleft, 
+            self.m2mright, 
+            self.m2mleft, 
+            self.m2mright, 
+            secondary=self.m2msecondary_ambig_fks
+        )
+
+    def test_determine_join_no_fks_m2m(self):
+        self._assert_raises_no_join(
+            relationships.JoinCondition,
+            "None", self.m2msecondary_no_fks,
+                    self.m2mleft, 
+                    self.m2mright, 
+                    self.m2mleft, 
+                    self.m2mright, 
+                    secondary=self.m2msecondary_no_fks
+        )
+
+    def _join_fixture_fks_ambig_m2m(self):
+        return relationships.JoinCondition(
+                    self.m2mleft, 
+                    self.m2mright, 
+                    self.m2mleft, 
+                    self.m2mright, 
+                    secondary=self.m2msecondary_ambig_fks,
+                    consider_as_foreign_keys=[
+                        self.m2msecondary_ambig_fks.c.lid1, 
+                        self.m2msecondary_ambig_fks.c.rid1]
+        )
+
+    def test_determine_join_w_fks_ambig_m2m(self):
+        joincond = self._join_fixture_fks_ambig_m2m()
+        self.assert_compile(
+                joincond.primaryjoin,
+                "m2mlft.id = m2msecondary_ambig_fks.lid1"
+        )
+        self.assert_compile(
+                joincond.secondaryjoin,
+                "m2mrgt.id = m2msecondary_ambig_fks.rid1"
+        )
+
+class AdaptedJoinTest(_JoinFixtures, fixtures.TestBase, AssertsCompiledSQL):
+    __dialect__ = 'default'
+
+    def test_join_targets_o2m_selfref(self):
+        joincond = self._join_fixture_o2m_selfref()
+        left = select([joincond.parent_selectable]).alias('pj')
+        pj, sj, sec, adapter, ds = joincond.join_targets(
+                                    left, 
+                                    joincond.child_selectable, 
+                                    True)
+        self.assert_compile(
+            pj, "pj.id = selfref.sid"
+        )
+
+        right = select([joincond.child_selectable]).alias('pj')
+        pj, sj, sec, adapter, ds = joincond.join_targets(
+                                    joincond.parent_selectable, 
+                                    right, 
+                                    True)
+        self.assert_compile(
+            pj, "selfref.id = pj.sid"
+        )
+
+
+    def test_join_targets_o2m_plain(self):
+        joincond = self._join_fixture_o2m()
+        pj, sj, sec, adapter, ds = joincond.join_targets(
+                                    joincond.parent_selectable, 
+                                    joincond.child_selectable, 
+                                    False)
+        self.assert_compile(
+            pj, "lft.id = rgt.lid"
+        )
+
+    def test_join_targets_o2m_left_aliased(self):
+        joincond = self._join_fixture_o2m()
+        left = select([joincond.parent_selectable]).alias('pj')
+        pj, sj, sec, adapter, ds = joincond.join_targets(
+                                    left, 
+                                    joincond.child_selectable, 
+                                    True)
+        self.assert_compile(
+            pj, "pj.id = rgt.lid"
+        )
+
+    def test_join_targets_o2m_right_aliased(self):
+        joincond = self._join_fixture_o2m()
+        right = select([joincond.child_selectable]).alias('pj')
+        pj, sj, sec, adapter, ds = joincond.join_targets(
+                                    joincond.parent_selectable, 
+                                    right, 
+                                    True)
+        self.assert_compile(
+            pj, "lft.id = pj.lid"
+        )
+
+    def test_join_targets_o2m_composite_selfref(self):
+        joincond = self._join_fixture_o2m_composite_selfref()
+        right = select([joincond.child_selectable]).alias('pj')
+        pj, sj, sec, adapter, ds = joincond.join_targets(
+                                    joincond.parent_selectable, 
+                                    right, 
+                                    True)
+        self.assert_compile(
+            pj, 
+            "pj.group_id = composite_selfref.group_id "
+            "AND composite_selfref.id = pj.parent_id"
+        )
+
+    def test_join_targets_m2o_composite_selfref(self):
+        joincond = self._join_fixture_m2o_composite_selfref()
+        right = select([joincond.child_selectable]).alias('pj')
+        pj, sj, sec, adapter, ds = joincond.join_targets(
+                                    joincond.parent_selectable, 
+                                    right, 
+                                    True)
+        self.assert_compile(
+            pj, 
+            "pj.group_id = composite_selfref.group_id "
+            "AND pj.id = composite_selfref.parent_id"
+        )
+
+class LazyClauseTest(_JoinFixtures, fixtures.TestBase, AssertsCompiledSQL):
+
+    def _test_lazy_clause_o2m(self):
+        joincond = self._join_fixture_o2m()
+        self.assert_compile(
+            relationships.create_lazy_clause(joincond),
+            ""
+        )
+
+    def _test_lazy_clause_o2m_reverse(self):
+        joincond = self._join_fixture_o2m()
+        self.assert_compile(
+            relationships.create_lazy_clause(joincond, 
+                                reverse_direction=True),
+            ""
+        )
+
diff --git a/test/orm/test_relationships.py b/test/orm/test_relationships.py
index d718c9d2d..6e610b0cf 100644
--- a/test/orm/test_relationships.py
+++ b/test/orm/test_relationships.py
@@ -7,11 +7,127 @@ from test.lib.schema import Table, Column
 from sqlalchemy.orm import mapper, relationship, relation, \
                     backref, create_session, configure_mappers, \
                     clear_mappers, sessionmaker, attributes,\
-                    Session, composite, column_property
-from test.lib.testing import eq_, startswith_
+                    Session, composite, column_property, foreign,\
+                    remote
+from sqlalchemy.orm.interfaces import ONETOMANY, MANYTOONE, MANYTOMANY
+from test.lib.testing import eq_, startswith_, AssertsCompiledSQL, is_
 from test.lib import fixtures
 from test.orm import _fixtures
+from sqlalchemy import exc
 
+class _RelationshipErrors(object):
+    def _assert_raises_no_relevant_fks(self, fn, expr, relname, 
+        primary, *arg, **kw):
+        assert_raises_message(
+            sa.exc.ArgumentError, 
+            "Could not locate any relevant foreign key columns "
+            "for %s join condition '%s' on relationship %s.  "
+            "Ensure that referencing columns are associated with "
+            "a ForeignKey or ForeignKeyConstraint, or are annotated "
+            r"in the join condition with the foreign\(\) annotation."
+            % (
+                primary, expr, relname
+            ),
+            fn, *arg, **kw
+        )
+
+    def _assert_raises_no_equality(self, fn, expr, relname, 
+        primary, *arg, **kw):
+        assert_raises_message(
+            sa.exc.ArgumentError, 
+            "Could not locate any simple equality expressions "
+            "involving locally mapped foreign key columns for %s join "
+            "condition '%s' on relationship %s.  "
+            "Ensure that referencing columns are associated with a "
+            "ForeignKey or ForeignKeyConstraint, or are annotated in "
+            r"the join condition with the foreign\(\) annotation. "
+            "To allow comparison operators other than '==', "
+            "the relationship can be marked as viewonly=True." % (
+                primary, expr, relname
+            ),
+            fn, *arg, **kw
+        )
+
+    def _assert_raises_ambig_join(self, fn, relname, secondary_arg,
+        *arg, **kw):
+        if secondary_arg is not None:
+            assert_raises_message(
+                exc.ArgumentError,
+                "Could not determine join condition between "
+                "parent/child tables on relationship %s - "
+                "there are multiple foreign key paths linking the "
+                "tables via secondary table '%s'.  "
+                "Specify the 'foreign_keys' argument, providing a list "
+                "of those columns which should be counted as "
+                "containing a foreign key reference from the "
+                "secondary table to each of the parent and child tables."
+                % (relname, secondary_arg),
+                fn, *arg, **kw)
+        else:
+            assert_raises_message(
+                exc.ArgumentError,
+                "Could not determine join "
+                "condition between parent/child tables on "
+                "relationship %s - there are multiple foreign key "
+                "paths linking the tables.  Specify the "
+                "'foreign_keys' argument, providing a list of those "
+                "columns which should be counted as containing a "
+                "foreign key reference to the parent table."
+                % (relname,),
+                fn, *arg, **kw)
+
+    def _assert_raises_no_join(self, fn, relname, secondary_arg,
+        *arg, **kw):
+        if secondary_arg is not None:
+            assert_raises_message(
+                exc.NoForeignKeysError,
+                "Could not determine join condition between "
+                "parent/child tables on relationship %s - "
+                "there are no foreign keys linking these tables "
+                "via secondary table '%s'.  "
+                "Ensure that referencing columns are associated with a ForeignKey "
+                "or ForeignKeyConstraint, or specify 'primaryjoin' and "
+                "'secondaryjoin' expressions"
+                % (relname, secondary_arg),
+                fn, *arg, **kw)
+        else:
+            assert_raises_message(
+                exc.NoForeignKeysError,
+                "Could not determine join condition between "
+                "parent/child tables on relationship %s - "
+                "there are no foreign keys linking these tables.  "
+                "Ensure that referencing columns are associated with a ForeignKey "
+                "or ForeignKeyConstraint, or specify a 'primaryjoin' "
+                "expression."
+                % (relname,),
+                fn, *arg, **kw)
+
+    def _assert_raises_ambiguous_direction(self, fn, relname, *arg, **kw):
+        assert_raises_message(
+            sa.exc.ArgumentError,
+            "Can't determine relationship"
+            " direction for relationship '%s' - foreign "
+            "key columns within the join condition are present "
+            "in both the parent and the child's mapped tables.  "
+            "Ensure that only those columns referring to a parent column "
+            r"are marked as foreign, either via the foreign\(\) annotation or "
+            "via the foreign_keys argument."
+            % relname,
+            fn, *arg, **kw
+        )
+
+    def _assert_raises_no_local_remote(self, fn, relname, *arg, **kw):
+        assert_raises_message(
+            sa.exc.ArgumentError,
+            "Relationship %s could not determine "
+            "any unambiguous local/remote column "
+            "pairs based on join condition and remote_side arguments.  "
+            r"Consider using the remote\(\) annotation to "
+            "accurately mark those elements of the join "
+            "condition that are on the remote side of the relationship." % relname,
+
+            fn, *arg, **kw
+        )
 
 class DependencyTwoParentTest(fixtures.MappedTest):
     """Test flush() when a mapper is dependent on multiple relationships"""
@@ -158,7 +274,8 @@ class CompositeSelfRefFKTest(fixtures.MappedTest):
     @classmethod
     def setup_classes(cls):
         class Company(cls.Basic):
-            pass
+            def __init__(self, name):
+                self.name = name
 
         class Employee(cls.Basic):
             def __init__(self, name, company, emp_id, reports_to=None):
@@ -185,8 +302,10 @@ class CompositeSelfRefFKTest(fixtures.MappedTest):
                     employee_t.c.company_id==employee_t.c.company_id
                 ),
                 remote_side=[employee_t.c.emp_id, employee_t.c.company_id],
-                foreign_keys=[employee_t.c.reports_to_id],
-                backref=backref('employees', foreign_keys=None))
+                foreign_keys=[employee_t.c.reports_to_id, employee_t.c.company_id],
+                backref=backref('employees', 
+                    foreign_keys=[employee_t.c.reports_to_id, 
+                                employee_t.c.company_id]))
         })
 
         self._test()
@@ -202,8 +321,10 @@ class CompositeSelfRefFKTest(fixtures.MappedTest):
             'company':relationship(Company, backref='employees'),
             'reports_to':relationship(Employee,
                 remote_side=[employee_t.c.emp_id, employee_t.c.company_id],
-                foreign_keys=[employee_t.c.reports_to_id],
-                backref=backref('employees', foreign_keys=None)
+                foreign_keys=[employee_t.c.reports_to_id, 
+                                    employee_t.c.company_id],
+                backref=backref('employees', foreign_keys=
+                    [employee_t.c.reports_to_id, employee_t.c.company_id])
                 )
         })
 
@@ -240,19 +361,70 @@ class CompositeSelfRefFKTest(fixtures.MappedTest):
                         (employee_t.c.reports_to_id, employee_t.c.emp_id),
                         (employee_t.c.company_id, employee_t.c.company_id)
                 ],
-                foreign_keys=[employee_t.c.reports_to_id],
-                backref=backref('employees', foreign_keys=None)
+                foreign_keys=[employee_t.c.reports_to_id, 
+                                    employee_t.c.company_id],
+                backref=backref('employees', foreign_keys=
+                    [employee_t.c.reports_to_id, employee_t.c.company_id])
+                )
+        })
+
+        self._test()
+
+    def test_annotated(self):
+        Employee, Company, employee_t, company_t = (self.classes.Employee,
+                                self.classes.Company,
+                                self.tables.employee_t,
+                                self.tables.company_t)
+
+        mapper(Company, company_t)
+        mapper(Employee, employee_t, properties= {
+            'company':relationship(Company, backref='employees'),
+            'reports_to':relationship(Employee,
+                primaryjoin=sa.and_(
+                    remote(employee_t.c.emp_id)==employee_t.c.reports_to_id,
+                    remote(employee_t.c.company_id)==employee_t.c.company_id
+                ),
+                backref=backref('employees')
                 )
         })
 
         self._test()
 
     def _test(self):
+        self._test_relationships()
+        sess = Session()
+        self._setup_data(sess)
+        self._test_lazy_relations(sess)
+        self._test_join_aliasing(sess)
+
+    def _test_relationships(self):
+        configure_mappers()
+        Employee = self.classes.Employee
+        employee_t = self.tables.employee_t
+        eq_(
+            set(Employee.employees.property.local_remote_pairs),
+            set([
+                (employee_t.c.company_id, employee_t.c.company_id),
+                (employee_t.c.emp_id, employee_t.c.reports_to_id),
+            ])
+        )
+        eq_(
+            Employee.employees.property.remote_side,
+            set([employee_t.c.company_id, employee_t.c.reports_to_id])
+        )
+        eq_(
+            set(Employee.reports_to.property.local_remote_pairs),
+            set([
+                (employee_t.c.company_id, employee_t.c.company_id),
+                (employee_t.c.reports_to_id, employee_t.c.emp_id),
+            ])
+        )
+
+    def _setup_data(self, sess):
         Employee, Company = self.classes.Employee, self.classes.Company
 
-        sess = create_session()
-        c1 = Company()
-        c2 = Company()
+        c1 = Company('c1')
+        c2 = Company('c2')
 
         e1 = Employee(u'emp1', c1, 1)
         e2 = Employee(u'emp2', c1, 2, e1)
@@ -263,10 +435,17 @@ class CompositeSelfRefFKTest(fixtures.MappedTest):
         e7 = Employee(u'emp7', c2, 3, e5)
 
         sess.add_all((c1, c2))
-        sess.flush()
-        sess.expunge_all()
+        sess.commit()
+        sess.close()
+
+    def _test_lazy_relations(self, sess):
+        Employee, Company = self.classes.Employee, self.classes.Company
+
+        c1 = sess.query(Company).filter_by(name='c1').one()
+        c2 = sess.query(Company).filter_by(name='c2').one()
+        e1 = sess.query(Employee).filter_by(name='emp1').one()
+        e5 = sess.query(Employee).filter_by(name='emp5').one()
 
-        test_c1 = sess.query(Company).get(c1.company_id)
         test_e1 = sess.query(Employee).get([c1.company_id, e1.emp_id])
         assert test_e1.name == 'emp1', test_e1.name
         test_e5 = sess.query(Employee).get([c2.company_id, e5.emp_id])
@@ -277,17 +456,63 @@ class CompositeSelfRefFKTest(fixtures.MappedTest):
         assert sess.query(Employee).\
                 get([c2.company_id, 3]).reports_to.name == 'emp5'
 
-        @testing.fails_if(lambda: True, "This will be fixed by #1401")
-        def go():
-            eq_(
-                [n for n, in sess.query(Employee.name).\
-                        join(Employee.reports_to, aliased=True).\
-                        filter_by(name='emp5').\
-                        reset_joinpoint().\
-                        order_by(Employee.name)],
-                ['emp6', 'emp7']
+    def _test_join_aliasing(self, sess):
+        Employee, Company = self.classes.Employee, self.classes.Company
+        eq_(
+            [n for n, in sess.query(Employee.name).\
+                    join(Employee.reports_to, aliased=True).\
+                    filter_by(name='emp5').\
+                    reset_joinpoint().\
+                    order_by(Employee.name)],
+            ['emp6', 'emp7']
+        )
+
+class CompositeJoinPartialFK(fixtures.MappedTest, AssertsCompiledSQL):
+    __dialect__ = 'default'
+    @classmethod
+    def define_tables(cls, metadata):
+        Table("parent", metadata,
+            Column('x', Integer, primary_key=True),
+            Column('y', Integer, primary_key=True),
+            Column('z', Integer),
+        )
+        Table("child", metadata,
+            Column('id', Integer, primary_key=True, 
+                        test_needs_autoincrement=True),
+            Column('x', Integer),
+            Column('y', Integer),
+            Column('z', Integer),
+            # note 'z' is not here
+            sa.ForeignKeyConstraint(
+                ["x", "y"],
+                ["parent.x", "parent.y"]
             )
-        go()
+        )
+    @classmethod
+    def setup_mappers(cls):
+        parent, child = cls.tables.parent, cls.tables.child
+        class Parent(cls.Comparable):
+            pass
+
+        class Child(cls.Comparable):
+            pass
+        mapper(Parent, parent, properties={
+            'children':relationship(Child, primaryjoin=and_(
+                parent.c.x==child.c.x,
+                parent.c.y==child.c.y,
+                parent.c.z==child.c.z,
+            ))
+        })
+        mapper(Child, child)
+
+    def test_joins_fully(self):
+        Parent, Child = self.classes.Parent, self.classes.Child
+        s = Session()
+        self.assert_compile(
+            Parent.children.property.strategy._lazywhere,
+            ":param_1 = child.x AND :param_2 = child.y AND :param_3 = child.z"
+        )
+
 
 class FKsAsPksTest(fixtures.MappedTest):
     """Syncrules on foreign keys that are also primary"""
@@ -768,7 +993,6 @@ class AmbiguousJoinInterpretedAsSelfRef(fixtures.MappedTest):
         subscriber_table = Table('subscriber', metadata,
            Column('id', Integer, primary_key=True, 
                             test_needs_autoincrement=True),
-           Column('dummy', String(10)) # to appease older sqlite version
           )
 
         address_table = Table('address',
@@ -803,7 +1027,6 @@ class AmbiguousJoinInterpretedAsSelfRef(fixtures.MappedTest):
     def test_mapping(self):
         Subscriber, Address = self.classes.Subscriber, self.classes.Address
 
-        from sqlalchemy.orm.interfaces import ONETOMANY, MANYTOONE
         sess = create_session()
         assert Subscriber.addresses.property.direction is ONETOMANY
         assert Address.customer.property.direction is MANYTOONE
@@ -1587,7 +1810,7 @@ class ViewOnlyRepeatedLocalColumn(fixtures.MappedTest):
         eq_(sess.query(Foo).filter_by(id=f2.id).one(),
             Foo(bars=[Bar(data='b3'), Bar(data='b4')]))
 
-class ViewOnlyComplexJoin(fixtures.MappedTest):
+class ViewOnlyComplexJoin(_RelationshipErrors, fixtures.MappedTest):
     """'viewonly' mappings with a complex join condition."""
 
     @classmethod
@@ -1671,9 +1894,7 @@ class ViewOnlyComplexJoin(fixtures.MappedTest):
             't1':relationship(T1),
             't3s':relationship(T3, secondary=t2tot3)})
         mapper(T3, t3)
-        assert_raises_message(sa.exc.ArgumentError,
-                                 "Specify remote_side argument",
-                                 sa.orm.configure_mappers)
+        self._assert_raises_no_local_remote(configure_mappers, "T1.t3s")
 
 
 class ExplicitLocalRemoteTest(fixtures.MappedTest):
@@ -1697,21 +1918,45 @@ class ExplicitLocalRemoteTest(fixtures.MappedTest):
         class T2(cls.Comparable):
             pass
 
-    def test_onetomany_funcfk(self):
+    def test_onetomany_funcfk_oldstyle(self):
         T2, T1, t2, t1 = (self.classes.T2,
                                 self.classes.T1,
                                 self.tables.t2,
                                 self.tables.t1)
 
-        # use a function within join condition.  but specifying
-        # local_remote_pairs overrides all parsing of the join condition.
+        # old _local_remote_pairs
         mapper(T1, t1, properties={
             't2s':relationship(T2,
                            primaryjoin=t1.c.id==sa.func.lower(t2.c.t1id),
                            _local_remote_pairs=[(t1.c.id, t2.c.t1id)],
-                           foreign_keys=[t2.c.t1id])})
+                           foreign_keys=[t2.c.t1id]
+                           )
+                          })
+        mapper(T2, t2)
+        self._test_onetomany()
+
+    def test_onetomany_funcfk_annotated(self):
+        T2, T1, t2, t1 = (self.classes.T2,
+                                self.classes.T1,
+                                self.tables.t2,
+                                self.tables.t1)
+
+        # use annotation
+        mapper(T1, t1, properties={
+            't2s':relationship(T2,
+                           primaryjoin=t1.c.id==
+                            foreign(sa.func.lower(t2.c.t1id)),
+                           )})
         mapper(T2, t2)
+        self._test_onetomany()
 
+    def _test_onetomany(self):
+        T2, T1, t2, t1 = (self.classes.T2,
+                                self.classes.T1,
+                                self.tables.t2,
+                                self.tables.t1)
+        is_(T1.t2s.property.direction, ONETOMANY)
+        eq_(T1.t2s.property.local_remote_pairs, [(t1.c.id, t2.c.t1id)])
         sess = create_session()
         a1 = T1(id='number1', data='a1')
         a2 = T1(id='number2', data='a2')
@@ -1910,8 +2155,135 @@ class InvalidRemoteSideTest(fixtures.MappedTest):
             "mean to set remote_side on the many-to-one side ?", 
             configure_mappers)
 
+class AmbiguousFKResolutionTest(_RelationshipErrors, fixtures.MappedTest):
+    @classmethod
+    def define_tables(cls, metadata):
+        Table("a", metadata,
+            Column('id', Integer, primary_key=True)
+        )
+        Table("b", metadata,
+            Column('id', Integer, primary_key=True),
+            Column('aid_1', Integer, ForeignKey('a.id')),
+            Column('aid_2', Integer, ForeignKey('a.id')),
+        )
+        Table("atob", metadata, 
+            Column('aid', Integer),
+            Column('bid', Integer),
+        )
+        Table("atob_ambiguous", metadata, 
+            Column('aid1', Integer, ForeignKey('a.id')),
+            Column('bid1', Integer, ForeignKey('b.id')),
+            Column('aid2', Integer, ForeignKey('a.id')),
+            Column('bid2', Integer, ForeignKey('b.id')),
+        )
+
+    @classmethod
+    def setup_classes(cls):
+        class A(cls.Basic):
+            pass
+        class B(cls.Basic):
+            pass
+
+    def test_ambiguous_fks_o2m(self):
+        A, B = self.classes.A, self.classes.B
+        a, b = self.tables.a, self.tables.b
+        mapper(A, a, properties={
+            'bs':relationship(B)
+        })
+        mapper(B, b)
+        self._assert_raises_ambig_join(
+            configure_mappers,
+            "A.bs",
+            None
+        )
+
+    def test_with_fks_o2m(self):
+        A, B = self.classes.A, self.classes.B
+        a, b = self.tables.a, self.tables.b
+        mapper(A, a, properties={
+            'bs':relationship(B, foreign_keys=b.c.aid_1)
+        })
+        mapper(B, b)
+        sa.orm.configure_mappers()
+        assert A.bs.property.primaryjoin.compare(
+            a.c.id==b.c.aid_1
+        )
+        eq_(
+            A.bs.property._calculated_foreign_keys,
+            set([b.c.aid_1])
+        )
+
+    def test_with_pj_o2m(self):
+        A, B = self.classes.A, self.classes.B
+        a, b = self.tables.a, self.tables.b
+        mapper(A, a, properties={
+            'bs':relationship(B, primaryjoin=a.c.id==b.c.aid_1)
+        })
+        mapper(B, b)
+        sa.orm.configure_mappers()
+        assert A.bs.property.primaryjoin.compare(
+            a.c.id==b.c.aid_1
+        )
+        eq_(
+            A.bs.property._calculated_foreign_keys,
+            set([b.c.aid_1])
+        )
+
+    def test_with_annotated_pj_o2m(self):
+        A, B = self.classes.A, self.classes.B
+        a, b = self.tables.a, self.tables.b
+        mapper(A, a, properties={
+            'bs':relationship(B, primaryjoin=a.c.id==foreign(b.c.aid_1))
+        })
+        mapper(B, b)
+        sa.orm.configure_mappers()
+        assert A.bs.property.primaryjoin.compare(
+            a.c.id==b.c.aid_1
+        )
+        eq_(
+            A.bs.property._calculated_foreign_keys,
+            set([b.c.aid_1])
+        )
+
+    def test_no_fks_m2m(self):
+        A, B = self.classes.A, self.classes.B
+        a, b, a_to_b = self.tables.a, self.tables.b, self.tables.atob
+        mapper(A, a, properties={
+            'bs':relationship(B, secondary=a_to_b)
+        })
+        mapper(B, b)
+        self._assert_raises_no_join(
+            sa.orm.configure_mappers,
+            "A.bs", a_to_b,
+        )
+
+    def test_ambiguous_fks_m2m(self):
+        A, B = self.classes.A, self.classes.B
+        a, b, a_to_b = self.tables.a, self.tables.b, self.tables.atob_ambiguous
+        mapper(A, a, properties={
+            'bs':relationship(B, secondary=a_to_b)
+        })
+        mapper(B, b)
+
+        self._assert_raises_ambig_join(
+            configure_mappers,
+            "A.bs",
+            "atob_ambiguous"
+        )
+
+
+    def test_with_fks_m2m(self):
+        A, B = self.classes.A, self.classes.B
+        a, b, a_to_b = self.tables.a, self.tables.b, self.tables.atob_ambiguous
+        mapper(A, a, properties={
+            'bs':relationship(B, secondary=a_to_b, 
+                        foreign_keys=[a_to_b.c.aid1, a_to_b.c.bid1])
+        })
+        mapper(B, b)
+        sa.orm.configure_mappers()
+
 
-class InvalidRelationshipEscalationTest(fixtures.MappedTest):
+class InvalidRelationshipEscalationTest(_RelationshipErrors, fixtures.MappedTest):
 
     @classmethod
     def define_tables(cls, metadata):
@@ -1936,6 +2308,7 @@ class InvalidRelationshipEscalationTest(fixtures.MappedTest):
         class Bar(cls.Basic):
             pass
 
+
     def test_no_join(self):
         bars, Foo, Bar, foos = (self.tables.bars,
                                 self.classes.Foo,
@@ -1946,10 +2319,9 @@ class InvalidRelationshipEscalationTest(fixtures.MappedTest):
             'bars':relationship(Bar)})
         mapper(Bar, bars)
 
-        assert_raises_message(
-            sa.exc.ArgumentError,
-            "Could not determine join condition between parent/child "
-            "tables on relationship", sa.orm.configure_mappers)
+        self._assert_raises_no_join(sa.orm.configure_mappers,
+            "Foo.bars", None
+        )
 
     def test_no_join_self_ref(self):
         bars, Foo, Bar, foos = (self.tables.bars,
@@ -1961,10 +2333,11 @@ class InvalidRelationshipEscalationTest(fixtures.MappedTest):
             'foos':relationship(Foo)})
         mapper(Bar, bars)
 
-        assert_raises_message(
-            sa.exc.ArgumentError,
-            "Could not determine join condition between parent/child "
-            "tables on relationship", sa.orm.configure_mappers)
+        self._assert_raises_no_join(
+            configure_mappers,
+            "Foo.foos",
+            None
+        )
 
     def test_no_equated(self):
         bars, Foo, Bar, foos = (self.tables.bars,
@@ -1977,11 +2350,10 @@ class InvalidRelationshipEscalationTest(fixtures.MappedTest):
                             primaryjoin=foos.c.id>bars.c.fid)})
         mapper(Bar, bars)
 
-        assert_raises_message(
-            sa.exc.ArgumentError,
-            "Could not determine relationship direction "
-            "for primaryjoin condition",
-            configure_mappers)
+        self._assert_raises_no_relevant_fks(
+            configure_mappers,
+            "foos.id > bars.fid", "Foo.bars", "primary"
+        )
 
     def test_no_equated_fks(self):
         bars, Foo, Bar, foos = (self.tables.bars,
@@ -1994,15 +2366,10 @@ class InvalidRelationshipEscalationTest(fixtures.MappedTest):
                             primaryjoin=foos.c.id>bars.c.fid,
                             foreign_keys=bars.c.fid)})
         mapper(Bar, bars)
-
-        assert_raises_message(
-            sa.exc.ArgumentError,
-            "Could not locate any foreign-key-equated, "
-            "locally mapped column pairs for primaryjoin "
-            "condition 'foos.id > bars.fid' on relationship "
-            "Foo.bars.  For more relaxed rules on join "
-            "conditions, the relationship may be marked as viewonly=True.",
-            sa.orm.configure_mappers)
+        self._assert_raises_no_equality(
+            sa.orm.configure_mappers,
+            "foos.id > bars.fid", "Foo.bars", "primary"
+        )
 
     def test_no_equated_wo_fks_works_on_relaxed(self):
         foos_with_fks, Foo, Bar, bars_with_fks, foos = (self.tables.foos_with_fks,
@@ -2026,19 +2393,12 @@ class InvalidRelationshipEscalationTest(fixtures.MappedTest):
                             )})
         mapper(Bar, bars_with_fks)
 
-        assert_raises_message(
-            sa.exc.ArgumentError,
-            "Could not locate any foreign-key-equated, locally mapped "
-             "column pairs for primaryjoin condition "
-             "'bars_with_fks.fid = foos_with_fks.id AND "
-             "foos_with_fks.id = foos.id' on relationship Foo.bars.  "
-             "Ensure that the referencing Column objects have a "
-             "ForeignKey present, or are otherwise part of a "
-             "ForeignKeyConstraint on their parent Table, or specify "
-             "the foreign_keys parameter to this relationship.  For "
-             "more relaxed rules on join conditions, the relationship "
-             "may be marked as viewonly=True.", 
-            sa.orm.configure_mappers)
+        self._assert_raises_no_equality(
+            sa.orm.configure_mappers,
+            "bars_with_fks.fid = foos_with_fks.id "
+            "AND foos_with_fks.id = foos.id", 
+            "Foo.bars", "primary"
+        )
 
     def test_ambiguous_fks(self):
         bars, Foo, Bar, foos = (self.tables.bars,
@@ -2052,20 +2412,10 @@ class InvalidRelationshipEscalationTest(fixtures.MappedTest):
                             foreign_keys=[foos.c.id, bars.c.fid])})
         mapper(Bar, bars)
 
-        assert_raises_message(sa.exc.ArgumentError,
-                              "Could not determine relationship "
-                              "direction for primaryjoin condition "
-                              "'foos.id = bars.fid', on relationship "
-                              "Foo.bars, using manual 'foreign_keys' "
-                              "setting.  Do the columns in "
-                              "'foreign_keys' represent all, and only, "
-                              "the 'foreign' columns in this join "
-                              r"condition\?  Does the mapped Table "
-                              "already have adequate ForeignKey and/or "
-                              "ForeignKeyConstraint objects "
-                              r"established \(in which case "
-                              r"'foreign_keys' is usually unnecessary\)\?"
-                              , sa.orm.configure_mappers)
+        self._assert_raises_ambiguous_direction(
+            sa.orm.configure_mappers,
+            "Foo.bars"
+        )
 
     def test_ambiguous_remoteside_o2m(self):
         bars, Foo, Bar, foos = (self.tables.bars,
@@ -2082,10 +2432,11 @@ class InvalidRelationshipEscalationTest(fixtures.MappedTest):
                             )})
         mapper(Bar, bars)
 
-        assert_raises_message(
-            sa.exc.ArgumentError, 
-                "could not determine any local/remote column pairs",
-                sa.orm.configure_mappers)
+        self._assert_raises_no_local_remote(
+            configure_mappers,
+            "Foo.bars",
+        )
+
 
     def test_ambiguous_remoteside_m2o(self):
         bars, Foo, Bar, foos = (self.tables.bars,
@@ -2102,13 +2453,13 @@ class InvalidRelationshipEscalationTest(fixtures.MappedTest):
                             )})
         mapper(Bar, bars)
 
-        assert_raises_message(
-            sa.exc.ArgumentError, 
-                "could not determine any local/remote column pairs",
-                sa.orm.configure_mappers)
+        self._assert_raises_no_local_remote(
+            configure_mappers,
+            "Foo.bars",
+        )
 
 
-    def test_no_equated_self_ref(self):
+    def test_no_equated_self_ref_no_fks(self):
         bars, Foo, Bar, foos = (self.tables.bars,
                                 self.classes.Foo,
                                 self.classes.Bar,
@@ -2119,13 +2470,12 @@ class InvalidRelationshipEscalationTest(fixtures.MappedTest):
                             primaryjoin=foos.c.id>foos.c.fid)})
         mapper(Bar, bars)
 
-        assert_raises_message(
-            sa.exc.ArgumentError,
-            "Could not determine relationship direction for primaryjoin "
-            "condition",
-            configure_mappers)
+        self._assert_raises_no_relevant_fks(configure_mappers, 
+                "foos.id > foos.fid", "Foo.foos", "primary"
+            )
 
-    def test_no_equated_self_ref(self):
+
+    def test_no_equated_self_ref_no_equality(self):
         bars, Foo, Bar, foos = (self.tables.bars,
                                 self.classes.Foo,
                                 self.classes.Bar,
@@ -2137,14 +2487,9 @@ class InvalidRelationshipEscalationTest(fixtures.MappedTest):
                             foreign_keys=[foos.c.fid])})
         mapper(Bar, bars)
 
-        assert_raises_message(
-            sa.exc.ArgumentError,
-            "Could not locate any foreign-key-equated, "
-            "locally mapped column pairs for primaryjoin "
-            "condition 'foos.id > foos.fid' on relationship "
-            "Foo.foos.  For more relaxed rules on join "
-            "conditions, the relationship may be marked as viewonly=True.",
-            sa.orm.configure_mappers)
+        self._assert_raises_no_equality(configure_mappers, 
+                "foos.id > foos.fid", "Foo.foos", "primary"
+            )
 
     def test_no_equated_viewonly(self):
         bars, Bar, bars_with_fks, foos_with_fks, Foo, foos = (self.tables.bars,
@@ -2160,10 +2505,10 @@ class InvalidRelationshipEscalationTest(fixtures.MappedTest):
                             viewonly=True)})
         mapper(Bar, bars)
 
-        assert_raises_message(sa.exc.ArgumentError,
-                              'Could not determine relationship '
-                              'direction for primaryjoin condition',
-                              sa.orm.configure_mappers)
+        self._assert_raises_no_relevant_fks(
+            sa.orm.configure_mappers,
+            "foos.id > bars.fid", "Foo.bars", "primary"
+        )
 
         sa.orm.clear_mappers()
         mapper(Foo, foos_with_fks, properties={
@@ -2187,15 +2532,10 @@ class InvalidRelationshipEscalationTest(fixtures.MappedTest):
                             viewonly=True)})
         mapper(Bar, bars)
 
-        assert_raises_message(sa.exc.ArgumentError,
-                              "Could not determine relationship "
-                              "direction for primaryjoin condition "
-                              "'foos.id > foos.fid', on relationship "
-                              "Foo.foos. Ensure that the referencing "
-                              "Column objects have a ForeignKey "
-                              "present, or are otherwise part of a "
-                              "ForeignKeyConstraint on their parent "
-                              "Table.", sa.orm.configure_mappers)
+        self._assert_raises_no_relevant_fks(
+            sa.orm.configure_mappers,
+            "foos.id > foos.fid", "Foo.foos", "primary"
+        )
 
         sa.orm.clear_mappers()
         mapper(Foo, foos_with_fks, properties={
@@ -2230,11 +2570,10 @@ class InvalidRelationshipEscalationTest(fixtures.MappedTest):
                             primaryjoin=foos.c.id==bars.c.fid)})
         mapper(Bar, bars)
 
-        assert_raises_message(
-            sa.exc.ArgumentError,
-            "Could not determine relationship direction for primaryjoin "
-            "condition",
-            configure_mappers)
+        self._assert_raises_no_relevant_fks(
+            configure_mappers,
+            "foos.id = bars.fid", "Foo.bars", "primary"
+        )
 
         sa.orm.clear_mappers()
         mapper(Foo, foos_with_fks, properties={
@@ -2250,11 +2589,10 @@ class InvalidRelationshipEscalationTest(fixtures.MappedTest):
             'foos':relationship(Foo,
                             primaryjoin=foos.c.id==foos.c.fid)})
 
-        assert_raises_message(
-            sa.exc.ArgumentError,
-            "Could not determine relationship direction for primaryjoin "
-            "condition",
-            configure_mappers)
+        self._assert_raises_no_relevant_fks(
+            configure_mappers,
+            "foos.id = foos.fid", "Foo.foos", "primary"
+        )
 
 
     def test_equated_self_ref_wrong_fks(self):
@@ -2267,14 +2605,13 @@ class InvalidRelationshipEscalationTest(fixtures.MappedTest):
                             primaryjoin=foos.c.id==foos.c.fid,
                             foreign_keys=[bars.c.id])})
 
-        assert_raises_message(
-            sa.exc.ArgumentError,
-            "Could not determine relationship direction for primaryjoin "
-            "condition",
-            configure_mappers)
+        self._assert_raises_no_relevant_fks(
+            configure_mappers,
+            "foos.id = foos.fid", "Foo.foos", "primary"
+        )
 
 
-class InvalidRelationshipEscalationTestM2M(fixtures.MappedTest):
+class InvalidRelationshipEscalationTestM2M(_RelationshipErrors, fixtures.MappedTest):
 
     @classmethod
     def define_tables(cls, metadata):
@@ -2317,10 +2654,11 @@ class InvalidRelationshipEscalationTestM2M(fixtures.MappedTest):
             'bars': relationship(Bar, secondary=foobars)})
         mapper(Bar, bars)
 
-        assert_raises_message(
-            sa.exc.ArgumentError,
-            "Could not determine join condition between parent/child tables "
-            "on relationship", sa.orm.configure_mappers)
+        self._assert_raises_no_join(
+            configure_mappers,
+            "Foo.bars", 
+            "foobars"
+        )
 
     def test_no_secondaryjoin(self):
         foobars, bars, Foo, Bar, foos = (self.tables.foobars,
@@ -2335,62 +2673,13 @@ class InvalidRelationshipEscalationTestM2M(fixtures.MappedTest):
                             primaryjoin=foos.c.id > foobars.c.fid)})
         mapper(Bar, bars)
 
-        assert_raises_message(
-            sa.exc.ArgumentError,
-            "Could not determine join condition between parent/child tables "
-            "on relationship",
-            sa.orm.configure_mappers)
-
-    def test_no_fks_warning_1(self):
-        foobars_with_many_columns, bars, Bar, foobars, Foo, foos = (self.tables.foobars_with_many_columns,
-                                self.tables.bars,
-                                self.classes.Bar,
-                                self.tables.foobars,
-                                self.classes.Foo,
-                                self.tables.foos)
-
-        mapper(Foo, foos, properties={
-            'bars': relationship(Bar, secondary=foobars, 
-                                primaryjoin=foos.c.id==foobars.c.fid,
-                                secondaryjoin=foobars.c.bid==bars.c.id)})
-        mapper(Bar, bars)
-
-        assert_raises_message(sa.exc.SAWarning,
-                              "No ForeignKey objects were present in "
-                              "secondary table 'foobars'.  Assumed "
-                              "referenced foreign key columns "
-                              "'foobars.bid', 'foobars.fid' for join "
-                              "condition 'foos.id = foobars.fid' on "
-                              "relationship Foo.bars",
-                              sa.orm.configure_mappers)
-
-        sa.orm.clear_mappers()
-        mapper(Foo, foos, properties={
-                        'bars': relationship(Bar, 
-                            secondary=foobars_with_many_columns, 
-                          primaryjoin=foos.c.id==
-                                            foobars_with_many_columns.c.fid,
-                          secondaryjoin=foobars_with_many_columns.c.bid==
-                                                    bars.c.id)})
-        mapper(Bar, bars)
+        self._assert_raises_no_join(
+            configure_mappers,
+            "Foo.bars", 
+            "foobars"
+        )
 
-        assert_raises_message(sa.exc.SAWarning,
-                              "No ForeignKey objects were present in "
-                              "secondary table 'foobars_with_many_colum"
-                              "ns'.  Assumed referenced foreign key "
-                              "columns 'foobars_with_many_columns.bid',"
-                              " 'foobars_with_many_columns.bid1', "
-                              "'foobars_with_many_columns.bid2', "
-                              "'foobars_with_many_columns.fid', "
-                              "'foobars_with_many_columns.fid1', "
-                              "'foobars_with_many_columns.fid2' for "
-                              "join condition 'foos.id = "
-                              "foobars_with_many_columns.fid' on "
-                              "relationship Foo.bars",
-                              sa.orm.configure_mappers)
-
-    @testing.emits_warning(r'No ForeignKey objects.*')
-    def test_no_fks_warning_2(self):
+    def test_no_fks(self):
         foobars_with_many_columns, bars, Bar, foobars, Foo, foos = (self.tables.foobars_with_many_columns,
                                 self.tables.bars,
                                 self.classes.Bar,
@@ -2448,11 +2737,11 @@ class InvalidRelationshipEscalationTestM2M(fixtures.MappedTest):
                              secondaryjoin=foobars.c.bid<=bars.c.id)})
         mapper(Bar, bars)
 
-        assert_raises_message(
-            sa.exc.ArgumentError,
-            "Could not determine relationship direction for "
-            "primaryjoin condition",
-            configure_mappers)
+        self._assert_raises_no_equality(
+                configure_mappers, 
+                'foos.id > foobars.fid',
+                "Foo.bars",
+                "primary")
 
         sa.orm.clear_mappers()
         mapper(Foo, foos, properties={
@@ -2461,18 +2750,11 @@ class InvalidRelationshipEscalationTestM2M(fixtures.MappedTest):
                              primaryjoin=foos.c.id > foobars_with_fks.c.fid,
                              secondaryjoin=foobars_with_fks.c.bid<=bars.c.id)})
         mapper(Bar, bars)
-        assert_raises_message(
-            sa.exc.ArgumentError,
-            r"Could not locate any foreign-key-equated, locally mapped "
-             "column pairs for primaryjoin condition 'foos.id > "
-             "foobars_with_fks.fid' on relationship Foo.bars.  Ensure "
-             "that the referencing Column objects have a ForeignKey "
-             "present, or are otherwise part of a ForeignKeyConstraint "
-             "on their parent Table, or specify the foreign_keys "
-             "parameter to this relationship.  For more relaxed "
-             "rules on join conditions, the relationship may be marked "
-             "as viewonly=True.", 
-            configure_mappers)
+        self._assert_raises_no_equality(
+                configure_mappers, 
+                'foos.id > foobars_with_fks.fid',
+                "Foo.bars",
+                "primary")
 
         sa.orm.clear_mappers()
         mapper(Foo, foos, properties={
@@ -2498,21 +2780,12 @@ class InvalidRelationshipEscalationTestM2M(fixtures.MappedTest):
                             secondaryjoin=foobars.c.bid <= bars.c.id,
                             foreign_keys=[foobars.c.fid])})
         mapper(Bar, bars)
-
-        assert_raises_message(sa.exc.ArgumentError,
-                              "Could not determine relationship "
-                              "direction for secondaryjoin condition "
-                              r"'foobars.bid \<\= bars.id', on "
-                              "relationship Foo.bars, using manual "
-                              "'foreign_keys' setting.  Do the columns "
-                              "in 'foreign_keys' represent all, and only, the "
-                              "'foreign' columns in this join "
-                              r"condition\?  Does the "
-                              "secondary Table already have adequate "
-                              "ForeignKey and/or ForeignKeyConstraint "
-                              r"objects established \(in which case "
-                              r"'foreign_keys' is usually unnecessary\)?"
-                              , sa.orm.configure_mappers)
+        self._assert_raises_no_relevant_fks(
+            configure_mappers,
+            "foobars.bid <= bars.id",
+            "Foo.bars",
+            "secondary"
+        )
 
     def test_no_equated_secondaryjoin(self):
         foobars, bars, Foo, Bar, foos = (self.tables.foobars,
@@ -2529,10 +2802,12 @@ class InvalidRelationshipEscalationTestM2M(fixtures.MappedTest):
                             foreign_keys=[foobars.c.fid, foobars.c.bid])})
         mapper(Bar, bars)
 
-        assert_raises_message(
-            sa.exc.ArgumentError,
-            "Could not locate any foreign-key-equated, locally mapped column pairs for "
-            "secondaryjoin condition", sa.orm.configure_mappers)
+        self._assert_raises_no_equality(
+            configure_mappers,
+            "foobars.bid <= bars.id",
+            "Foo.bars",
+            "secondary"
+        )
 
 class ActiveHistoryFlagTest(_fixtures.FixtureTest):
     run_inserts = None
diff --git a/test/sql/test_generative.py b/test/sql/test_generative.py
index 98e783ede..29b7cd482 100644
--- a/test/sql/test_generative.py
+++ b/test/sql/test_generative.py
@@ -1,5 +1,5 @@
 from sqlalchemy import *
-from sqlalchemy.sql import table, column, ClauseElement
+from sqlalchemy.sql import table, column, ClauseElement, operators
 from sqlalchemy.sql.expression import  _clone, _from_objects
 from test.lib import *
 from sqlalchemy.sql.visitors import *
@@ -166,6 +166,102 @@ class TraversalTest(fixtures.TestBase, AssertsExecutionResults):
         s = set(ClauseVisitor().iterate(bin))
         assert set(ClauseVisitor().iterate(bin)) == set([foo, bar, bin])
 
+class BinaryEndpointTraversalTest(fixtures.TestBase):
+    """test the special binary product visit"""
+
+    def _assert_traversal(self, expr, expected):
+        canary = []
+        def visit(binary, l, r):
+            canary.append((binary.operator, l, r))
+            print binary.operator, l, r
+        sql_util.visit_binary_product(visit, expr)
+        eq_(
+            canary, expected
+        )
+
+    def test_basic(self):
+        a, b = column("a"), column("b")
+        self._assert_traversal(
+            a == b,
+            [
+                (operators.eq, a, b)
+            ]
+        )
+
+    def test_with_tuples(self):
+        a, b, c, d, b1, b1a, b1b, e, f = (
+            column("a"),
+            column("b"),
+            column("c"),
+            column("d"),
+            column("b1"),
+            column("b1a"),
+            column("b1b"),
+            column("e"),
+            column("f")
+        )
+        expr = tuple_(
+            a, b, b1==tuple_(b1a, b1b == d), c
+        ) > tuple_(
+                func.go(e + f)
+            )
+        self._assert_traversal(
+            expr,
+            [
+                (operators.gt, a, e),
+                (operators.gt, a, f),
+                (operators.gt, b, e),
+                (operators.gt, b, f),
+                (operators.eq, b1, b1a),
+                (operators.eq, b1b, d),
+                (operators.gt, c, e),
+                (operators.gt, c, f)
+            ]
+        )
+
+    def test_composed(self):
+        a, b, e, f, q, j, r = (
+            column("a"),
+            column("b"),
+            column("e"),
+            column("f"),
+            column("q"),
+            column("j"),
+            column("r"),
+        )
+        expr = and_(
+            (a + b) == q + func.sum(e + f),
+            and_(
+                j == r,
+                f == q
+            )
+        )
+        self._assert_traversal(
+            expr,
+            [
+                (operators.eq, a, q),
+                (operators.eq, a, e),
+                (operators.eq, a, f),
+                (operators.eq, b, q),
+                (operators.eq, b, e),
+                (operators.eq, b, f),
+                (operators.eq, j, r),
+                (operators.eq, f, q),
+            ]
+        )
+
+    def test_subquery(self):
+        a, b, c = column("a"), column("b"), column("c")
+        subq = select([c]).where(c == a).as_scalar()
+        expr = and_(a == b, b == subq)
+        self._assert_traversal(
+            expr,
+            [
+                (operators.eq, a, b),
+                (operators.eq, b, subq),
+            ]
+        )
+
 class ClauseTest(fixtures.TestBase, AssertsCompiledSQL):
     """test copy-in-place behavior of various ClauseElements."""
 
diff --git a/test/sql/test_selectable.py b/test/sql/test_selectable.py
index bbb9131a5..dde832e7d 100644
--- a/test/sql/test_selectable.py
+++ b/test/sql/test_selectable.py
@@ -1023,6 +1023,25 @@ class AnnotationsTest(fixtures.TestBase):
             annot = obj._annotate({})
             eq_(set([obj]), set([annot]))
 
+    def test_compare(self):
+        t = table('t', column('x'), column('y'))
+        x_a = t.c.x._annotate({})
+        assert t.c.x.compare(x_a)
+        assert x_a.compare(t.c.x)
+        assert not x_a.compare(t.c.y)
+        assert not t.c.y.compare(x_a)
+        assert (t.c.x == 5).compare(x_a == 5)
+        assert not (t.c.y == 5).compare(x_a == 5)
+
+        s = select([t])
+        x_p = s.c.x
+        assert not x_a.compare(x_p)
+        assert not t.c.x.compare(x_p)
+        x_p_a = x_p._annotate({})
+        assert x_p_a.compare(x_p)
+        assert x_p.compare(x_p_a)
+        assert not x_p_a.compare(x_a)
+
     def test_custom_constructions(self):
         from sqlalchemy.schema import Column
         class MyColumn(Column):
@@ -1132,13 +1151,18 @@ class AnnotationsTest(fixtures.TestBase):
         assert b2.left is not bin.left 
         assert b3.left is not b2.left is not bin.left
         assert b4.left is bin.left  # since column is immutable
-        assert b4.right is not bin.right is not b2.right is not b3.right
+        # deannotate copies the element
+        assert bin.right is not b2.right is not b3.right is not b4.right
 
     def test_annotate_unique_traversal(self):
         """test that items are copied only once during
         annotate, deannotate traversal
 
-        #2453
+        #2453 - however note this was modified by
+        #1401, and it's likely that re49563072578
+        is helping us with the str() comparison
+        case now, as deannotate is making 
+        clones again in some cases.
         """
         table1 = table('table1', column('x'))
         table2 = table('table2', column('y'))
@@ -1146,21 +1170,81 @@ class AnnotationsTest(fixtures.TestBase):
         s = select([a1.c.x]).select_from(
                 a1.join(table2, a1.c.x==table2.c.y)
             )
-
         for sel in (
             sql_util._deep_deannotate(s),
-            sql_util._deep_annotate(s, {'foo':'bar'}),
             visitors.cloned_traverse(s, {}, {}),
             visitors.replacement_traverse(s, {}, lambda x:None)
         ):
             # the columns clause isn't changed at all
             assert sel._raw_columns[0].table is a1
-            # the from objects are internally consistent,
-            # i.e. the Alias at position 0 is the same
-            # Alias in the Join object in position 1
             assert sel._froms[0] is sel._froms[1].left
+
+            eq_(str(s), str(sel))
+
+        # when we are modifying annotations sets only
+        # partially, each element is copied unconditionally
+        # when encountered.
+        for sel in (
+            sql_util._deep_deannotate(s, {"foo":"bar"}),
+            sql_util._deep_annotate(s, {'foo':'bar'}),
+        ):
+            assert sel._froms[0] is not sel._froms[1].left
+
+            # but things still work out due to
+            # re49563072578
             eq_(str(s), str(sel))
 
+
+    def test_annotate_varied_annot_same_col(self):
+        """test two instances of the same column with different annotations
+        preserving them when deep_annotate is run on them.
+        
+        """
+        t1 = table('table1', column("col1"), column("col2"))
+        s = select([t1.c.col1._annotate({"foo":"bar"})])
+        s2 = select([t1.c.col1._annotate({"bat":"hoho"})])
+        s3 = s.union(s2)
+        sel = sql_util._deep_annotate(s3, {"new":"thing"})
+
+        eq_(
+            sel.selects[0]._raw_columns[0]._annotations,
+            {"foo":"bar", "new":"thing"}
+        )
+
+        eq_(
+            sel.selects[1]._raw_columns[0]._annotations,
+            {"bat":"hoho", "new":"thing"}
+        )
+
+    def test_deannotate_2(self):
+        table1 = table('table1', column("col1"), column("col2"))
+        j = table1.c.col1._annotate({"remote":True}) == \
+                table1.c.col2._annotate({"local":True})
+        j2 = sql_util._deep_deannotate(j)
+        eq_(
+            j.left._annotations, {"remote":True}
+        )
+        eq_(
+            j2.left._annotations, {}
+        )
+
+    def test_deannotate_3(self):
+        table1 = table('table1', column("col1"), column("col2"), 
+                            column("col3"), column("col4"))
+        j = and_(
+                table1.c.col1._annotate({"remote":True})==
+                table1.c.col2._annotate({"local":True}), 
+                table1.c.col3._annotate({"remote":True})==
+                table1.c.col4._annotate({"local":True})
+        )
+        j2 = sql_util._deep_deannotate(j)
+        eq_(
+            j.clauses[0].left._annotations, {"remote":True}
+        )
+        eq_(
+            j2.clauses[0].left._annotations, {}
+        )
+
     def test_annotate_fromlist_preservation(self):
         """test the FROM list in select still works
         even when multiple annotate runs have created