- 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, 1223 insertions, 657 deletions

diff --git a/lib/sqlalchemy/exc.py b/lib/sqlalchemy/exc.py
index 91ffc2811..f28bd8a07 100644
--- a/lib/sqlalchemy/exc.py
+++ b/lib/sqlalchemy/exc.py
@@ -25,6 +25,13 @@ class ArgumentError(SQLAlchemyError):
 
     """
 
+class NoForeignKeysError(ArgumentError):
+    """Raised when no foreign keys can be located between two selectables
+    during a join."""
+
+class AmbiguousForeignKeysError(ArgumentError):
+    """Raised when more than one foreign key matching can be located
+    between two selectables during a join."""
 
 class CircularDependencyError(SQLAlchemyError):
     """Raised by topological sorts when a circular dependency is detected.
diff --git a/lib/sqlalchemy/ext/declarative.py b/lib/sqlalchemy/ext/declarative.py
index faf575da1..07a2a3b95 100755
--- a/lib/sqlalchemy/ext/declarative.py
+++ b/lib/sqlalchemy/ext/declarative.py
@@ -1398,6 +1398,10 @@ class _GetTable(object):
 def _deferred_relationship(cls, prop):
     def resolve_arg(arg):
         import sqlalchemy
+        from sqlalchemy.orm import foreign, remote
+
+        fallback = sqlalchemy.__dict__.copy()
+        fallback.update({'foreign':foreign, 'remote':remote})
 
         def access_cls(key):
             if key in cls._decl_class_registry:
@@ -1407,7 +1411,7 @@ def _deferred_relationship(cls, prop):
             elif key in cls.metadata._schemas:
                 return _GetTable(key, cls.metadata)
             else:
-                return sqlalchemy.__dict__[key]
+                return fallback[key]
 
         d = util.PopulateDict(access_cls)
         def return_cls():
diff --git a/lib/sqlalchemy/orm/__init__.py b/lib/sqlalchemy/orm/__init__.py
index ca4099d68..1813f57d8 100644
--- a/lib/sqlalchemy/orm/__init__.py
+++ b/lib/sqlalchemy/orm/__init__.py
@@ -44,6 +44,11 @@ from sqlalchemy.orm.properties import (
      PropertyLoader,
      SynonymProperty,
      )
+from sqlalchemy.orm.relationships import (
+    foreign,
+    remote,
+    remote_foreign
+)
 from sqlalchemy.orm import mapper as mapperlib
 from sqlalchemy.orm.mapper import reconstructor, validates
 from sqlalchemy.orm import strategies
@@ -81,6 +86,7 @@ __all__ = (
     'dynamic_loader',
     'eagerload',
     'eagerload_all',
+    'foreign',
     'immediateload',
     'join',
     'joinedload',
@@ -96,6 +102,8 @@ __all__ = (
     'reconstructor',
     'relationship',
     'relation',
+    'remote',
+    'remote_foreign',
     'scoped_session',
     'sessionmaker',
     'subqueryload',
diff --git a/lib/sqlalchemy/orm/properties.py b/lib/sqlalchemy/orm/properties.py
index 74ccf0157..424795ee4 100644
--- a/lib/sqlalchemy/orm/properties.py
+++ b/lib/sqlalchemy/orm/properties.py
@@ -14,11 +14,11 @@ mapped attributes.
 from sqlalchemy import sql, util, log, exc as sa_exc
 from sqlalchemy.sql.util import ClauseAdapter, criterion_as_pairs, \
     join_condition, _shallow_annotate
-from sqlalchemy.sql import operators, expression
+from sqlalchemy.sql import operators, expression, visitors
 from sqlalchemy.orm import attributes, dependency, mapper, \
-    object_mapper, strategies, configure_mappers
+    object_mapper, strategies, configure_mappers, relationships
 from sqlalchemy.orm.util import CascadeOptions, _class_to_mapper, \
-    _orm_annotate, _orm_deannotate
+    _orm_annotate, _orm_deannotate, _orm_full_deannotate
 
 from sqlalchemy.orm.interfaces import MANYTOMANY, MANYTOONE, \
     MapperProperty, ONETOMANY, PropComparator, StrategizedProperty
@@ -33,9 +33,9 @@ from descriptor_props import CompositeProperty, SynonymProperty, \
 
 class ColumnProperty(StrategizedProperty):
     """Describes an object attribute that corresponds to a table column.
-    
+
     Public constructor is the :func:`.orm.column_property` function.
-    
+
     """
 
     def __init__(self, *columns, **kwargs):
@@ -62,7 +62,7 @@ class ColumnProperty(StrategizedProperty):
 
         """
         self._orig_columns = [expression._labeled(c) for c in columns]
-        self.columns = [expression._labeled(_orm_deannotate(c)) 
+        self.columns = [expression._labeled(_orm_full_deannotate(c)) 
                             for c in columns]
         self.group = kwargs.pop('group', None)
         self.deferred = kwargs.pop('deferred', False)
@@ -177,13 +177,13 @@ log.class_logger(ColumnProperty)
 class RelationshipProperty(StrategizedProperty):
     """Describes an object property that holds a single item or list
     of items that correspond to a related database table.
-    
+
     Public constructor is the :func:`.orm.relationship` function.
-    
+
     Of note here is the :class:`.RelationshipProperty.Comparator`
     class, which implements comparison operations for scalar-
     and collection-referencing mapped attributes.
-    
+
     """
 
     strategy_wildcard_key = 'relationship:*'
@@ -293,7 +293,7 @@ class RelationshipProperty(StrategizedProperty):
         def __init__(self, prop, mapper, of_type=None, adapter=None):
             """Construction of :class:`.RelationshipProperty.Comparator`
             is internal to the ORM's attribute mechanics.
-            
+
             """
             self.prop = prop
             self.mapper = mapper
@@ -332,10 +332,10 @@ class RelationshipProperty(StrategizedProperty):
         def of_type(self, cls):
             """Produce a construct that represents a particular 'subtype' of
             attribute for the parent class.
-            
+
             Currently this is usable in conjunction with :meth:`.Query.join`
             and :meth:`.Query.outerjoin`.
-            
+
             """
             return RelationshipProperty.Comparator(
                                         self.property, 
@@ -345,7 +345,7 @@ class RelationshipProperty(StrategizedProperty):
         def in_(self, other):
             """Produce an IN clause - this is not implemented 
             for :func:`~.orm.relationship`-based attributes at this time.
-            
+
             """
             raise NotImplementedError('in_() not yet supported for '
                     'relationships.  For a simple many-to-one, use '
@@ -362,15 +362,15 @@ class RelationshipProperty(StrategizedProperty):
 
             this will typically produce a
             clause such as::
-  
+
               mytable.related_id == <some id>
-  
+
             Where ``<some id>`` is the primary key of the given 
             object.
-  
+
             The ``==`` operator provides partial functionality for non-
             many-to-one comparisons:
-  
+
             * Comparisons against collections are not supported.
               Use :meth:`~.RelationshipProperty.Comparator.contains`.
             * Compared to a scalar one-to-many, will produce a 
@@ -445,6 +445,7 @@ class RelationshipProperty(StrategizedProperty):
             else:
                 j = _orm_annotate(pj, exclude=self.property.remote_side)
 
+            # MARKMARK
             if criterion is not None and target_adapter:
                 # limit this adapter to annotated only?
                 criterion = target_adapter.traverse(criterion)
@@ -465,42 +466,42 @@ class RelationshipProperty(StrategizedProperty):
         def any(self, criterion=None, **kwargs):
             """Produce an expression that tests a collection against
             particular criterion, using EXISTS.
-            
+
             An expression like::
-            
+
                 session.query(MyClass).filter(
                     MyClass.somereference.any(SomeRelated.x==2)
                 )
-                
-                
+
+
             Will produce a query like::
-            
+
                 SELECT * FROM my_table WHERE
                 EXISTS (SELECT 1 FROM related WHERE related.my_id=my_table.id 
                 AND related.x=2)
-                
+
             Because :meth:`~.RelationshipProperty.Comparator.any` uses
             a correlated subquery, its performance is not nearly as
             good when compared against large target tables as that of
             using a join.
-            
+
             :meth:`~.RelationshipProperty.Comparator.any` is particularly
             useful for testing for empty collections::
-            
+
                 session.query(MyClass).filter(
                     ~MyClass.somereference.any()
                 )
-            
+
             will produce::
-            
+
                 SELECT * FROM my_table WHERE
                 NOT EXISTS (SELECT 1 FROM related WHERE related.my_id=my_table.id)
-                
+
             :meth:`~.RelationshipProperty.Comparator.any` is only
             valid for collections, i.e. a :func:`.relationship`
             that has ``uselist=True``.  For scalar references,
             use :meth:`~.RelationshipProperty.Comparator.has`.
-            
+
             """
             if not self.property.uselist:
                 raise sa_exc.InvalidRequestError(
@@ -515,14 +516,14 @@ class RelationshipProperty(StrategizedProperty):
             particular criterion, using EXISTS.
 
             An expression like::
-            
+
                 session.query(MyClass).filter(
                     MyClass.somereference.has(SomeRelated.x==2)
                 )
-                
-                
+
+
             Will produce a query like::
-            
+
                 SELECT * FROM my_table WHERE
                 EXISTS (SELECT 1 FROM related WHERE related.id==my_table.related_id
                 AND related.x=2)
@@ -531,12 +532,12 @@ class RelationshipProperty(StrategizedProperty):
             a correlated subquery, its performance is not nearly as
             good when compared against large target tables as that of
             using a join.
-            
+
             :meth:`~.RelationshipProperty.Comparator.has` is only
             valid for scalar references, i.e. a :func:`.relationship`
             that has ``uselist=False``.  For collection references,
             use :meth:`~.RelationshipProperty.Comparator.any`.
-            
+
             """
             if self.property.uselist:
                 raise sa_exc.InvalidRequestError(
@@ -547,44 +548,44 @@ class RelationshipProperty(StrategizedProperty):
         def contains(self, other, **kwargs):
             """Return a simple expression that tests a collection for 
             containment of a particular item.
-            
+
             :meth:`~.RelationshipProperty.Comparator.contains` is
             only valid for a collection, i.e. a
             :func:`~.orm.relationship` that implements
             one-to-many or many-to-many with ``uselist=True``.
-            
+
             When used in a simple one-to-many context, an 
             expression like::
-            
+
                 MyClass.contains(other)
-                
+
             Produces a clause like::
-            
+
                 mytable.id == <some id>
-                
+
             Where ``<some id>`` is the value of the foreign key
             attribute on ``other`` which refers to the primary
             key of its parent object. From this it follows that
             :meth:`~.RelationshipProperty.Comparator.contains` is
             very useful when used with simple one-to-many
             operations.
-            
+
             For many-to-many operations, the behavior of
             :meth:`~.RelationshipProperty.Comparator.contains`
             has more caveats. The association table will be
             rendered in the statement, producing an "implicit"
             join, that is, includes multiple tables in the FROM
             clause which are equated in the WHERE clause::
-            
+
                 query(MyClass).filter(MyClass.contains(other))
-                
+
             Produces a query like::
-            
+
                 SELECT * FROM my_table, my_association_table AS
                 my_association_table_1 WHERE
                 my_table.id = my_association_table_1.parent_id
                 AND my_association_table_1.child_id = <some id>
-                
+
             Where ``<some id>`` would be the primary key of
             ``other``. From the above, it is clear that
             :meth:`~.RelationshipProperty.Comparator.contains`
@@ -598,7 +599,7 @@ class RelationshipProperty(StrategizedProperty):
             a less-performant alternative using EXISTS, or refer
             to :meth:`.Query.outerjoin` as well as :ref:`ormtutorial_joins`
             for more details on constructing outer joins.
-            
+
             """
             if not self.property.uselist:
                 raise sa_exc.InvalidRequestError(
@@ -649,19 +650,19 @@ class RelationshipProperty(StrategizedProperty):
             """Implement the ``!=`` operator.
 
             In a many-to-one context, such as::
-  
+
               MyClass.some_prop != <some object>
-  
+
             This will typically produce a clause such as::
-  
+
               mytable.related_id != <some id>
-  
+
             Where ``<some id>`` is the primary key of the
             given object.
-  
+
             The ``!=`` operator provides partial functionality for non-
             many-to-one comparisons:
-  
+
             * Comparisons against collections are not supported.
               Use
               :meth:`~.RelationshipProperty.Comparator.contains`
@@ -682,7 +683,7 @@ class RelationshipProperty(StrategizedProperty):
               membership tests.
             * Comparisons against ``None`` given in a one-to-many
               or many-to-many context produce an EXISTS clause.
-                
+
             """
             if isinstance(other, (NoneType, expression._Null)):
                 if self.property.direction == MANYTOONE:
@@ -880,9 +881,9 @@ class RelationshipProperty(StrategizedProperty):
     def mapper(self):
         """Return the targeted :class:`.Mapper` for this 
         :class:`.RelationshipProperty`.
-        
+
         This is a lazy-initializing static attribute.
-        
+
         """
         if isinstance(self.argument, type):
             mapper_ = mapper.class_mapper(self.argument,
@@ -914,58 +915,25 @@ class RelationshipProperty(StrategizedProperty):
     def do_init(self):
         self._check_conflicts()
         self._process_dependent_arguments()
-        self._determine_joins()
-        self._determine_synchronize_pairs()
-        self._determine_direction()
-        self._determine_local_remote_pairs()
+        self._setup_join_conditions()
+        self._check_cascade_settings()
         self._post_init()
         self._generate_backref()
         super(RelationshipProperty, self).do_init()
 
-    def _check_conflicts(self):
-        """Test that this relationship is legal, warn about 
-        inheritance conflicts."""
-
-        if not self.is_primary() \
-            and not mapper.class_mapper(
-                                self.parent.class_,
-                                compile=False).has_property(self.key):
-            raise sa_exc.ArgumentError("Attempting to assign a new "
-                    "relationship '%s' to a non-primary mapper on "
-                    "class '%s'.  New relationships can only be added "
-                    "to the primary mapper, i.e. the very first mapper "
-                    "created for class '%s' " % (self.key,
-                    self.parent.class_.__name__,
-                    self.parent.class_.__name__))
-
-        # check for conflicting relationship() on superclass
-        if not self.parent.concrete:
-            for inheriting in self.parent.iterate_to_root():
-                if inheriting is not self.parent \
-                    and inheriting.has_property(self.key):
-                    util.warn("Warning: relationship '%s' on mapper "
-                              "'%s' supersedes the same relationship "
-                              "on inherited mapper '%s'; this can "
-                              "cause dependency issues during flush"
-                              % (self.key, self.parent, inheriting))
-
     def _process_dependent_arguments(self):
         """Convert incoming configuration arguments to their 
         proper form.
-        
+
         Callables are resolved, ORM annotations removed.
-        
+
         """
         # accept callables for other attributes which may require
         # deferred initialization.  This technique is used
         # by declarative "string configs" and some recipes.
         for attr in (
-            'order_by',
-            'primaryjoin',
-            'secondaryjoin',
-            'secondary',
-            '_user_defined_foreign_keys',
-            'remote_side',
+            'order_by', 'primaryjoin', 'secondaryjoin',
+            'secondary', '_user_defined_foreign_keys', 'remote_side',
             ):
             attr_value = getattr(self, attr)
             if util.callable(attr_value):
@@ -1008,284 +976,64 @@ class RelationshipProperty(StrategizedProperty):
                             (self.key, self.parent.class_)
                         )
 
-    def _determine_joins(self):
-        """Determine the 'primaryjoin' and 'secondaryjoin' attributes,
-        if not passed to the constructor already.
-        
-        This is based on analysis of the foreign key relationships
-        between the parent and target mapped selectables.
-        
-        """
-        if self.secondaryjoin is not None and self.secondary is None:
-            raise sa_exc.ArgumentError("Property '" + self.key
-                    + "' specified with secondary join condition but "
-                    "no secondary argument")
-
-        # if join conditions were not specified, figure them out based
-        # on foreign keys
-
-        def _search_for_join(mapper, table):
-            # find a join between the given mapper's mapped table and
-            # the given table. will try the mapper's local table first
-            # for more specificity, then if not found will try the more
-            # general mapped table, which in the case of inheritance is
-            # a join.
-            return join_condition(mapper.mapped_table, table, 
-                                        a_subset=mapper.local_table)
-
-        try:
-            if self.secondary is not None:
-                if self.secondaryjoin is None:
-                    self.secondaryjoin = _search_for_join(self.mapper,
-                            self.secondary)
-                if self.primaryjoin is None:
-                    self.primaryjoin = _search_for_join(self.parent,
-                            self.secondary)
-            else:
-                if self.primaryjoin is None:
-                    self.primaryjoin = _search_for_join(self.parent,
-                            self.target)
-        except sa_exc.ArgumentError, e:
-            raise sa_exc.ArgumentError("Could not determine join "
-                    "condition between parent/child tables on "
-                    "relationship %s.  Specify a 'primaryjoin' "
-                    "expression.  If 'secondary' is present, "
-                    "'secondaryjoin' is needed as well."
-                    % self)
-
-    def _columns_are_mapped(self, *cols):
-        """Return True if all columns in the given collection are 
-        mapped by the tables referenced by this :class:`.Relationship`.
-        
-        """
-        for c in cols:
-            if self.secondary is not None \
-                and self.secondary.c.contains_column(c):
-                continue
-            if not self.parent.mapped_table.c.contains_column(c) and \
-                not self.target.c.contains_column(c):
-                return False
-        return True
-
-    def _sync_pairs_from_join(self, join_condition, primary):
-        """Determine a list of "source"/"destination" column pairs
-        based on the given join condition, as well as the
-        foreign keys argument.
-        
-        "source" would be a column referenced by a foreign key,
-        and "destination" would be the column who has a foreign key
-        reference to "source".
-        
-        """
-
-        fks = self._user_defined_foreign_keys
-        # locate pairs
-        eq_pairs = criterion_as_pairs(join_condition,
-                consider_as_foreign_keys=fks,
-                any_operator=self.viewonly)
-
-        # couldn't find any fks, but we have 
-        # "secondary" - assume the "secondary" columns
-        # are the fks
-        if not eq_pairs and \
-                self.secondary is not None and \
-                not fks:
-            fks = set(self.secondary.c)
-            eq_pairs = criterion_as_pairs(join_condition,
-                    consider_as_foreign_keys=fks,
-                    any_operator=self.viewonly)
-
-            if eq_pairs:
-                util.warn("No ForeignKey objects were present "
-                            "in secondary table '%s'.  Assumed referenced "
-                            "foreign key columns %s for join condition '%s' "
-                            "on relationship %s" % (
-                            self.secondary.description,
-                            ", ".join(sorted(["'%s'" % col for col in fks])),
-                            join_condition,
-                            self
-                        ))
-
-        # Filter out just to columns that are mapped.
-        # If viewonly, allow pairs where the FK col
-        # was part of "foreign keys" - the column it references
-        # may be in an un-mapped table - see 
-        # test.orm.test_relationships.ViewOnlyComplexJoin.test_basic
-        # for an example of this.
-        eq_pairs = [(l, r) for (l, r) in eq_pairs
-                    if self._columns_are_mapped(l, r)
-                    or self.viewonly and 
-                    r in fks]
-
-        if eq_pairs:
-            return eq_pairs
-
-        # from here below is just determining the best error message
-        # to report.  Check for a join condition using any operator 
-        # (not just ==), perhaps they need to turn on "viewonly=True".
-        if not self.viewonly and criterion_as_pairs(join_condition,
-                consider_as_foreign_keys=self._user_defined_foreign_keys,
-                any_operator=True):
-
-            err = "Could not locate any "\
-                    "foreign-key-equated, locally mapped column "\
-                    "pairs for %s "\
-                    "condition '%s' on relationship %s." % (
-                        primary and 'primaryjoin' or 'secondaryjoin', 
-                        join_condition, 
-                        self
-                    )
-
-            if not self._user_defined_foreign_keys:
-                err += "  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."
-
-            err += "  For more "\
-                    "relaxed rules on join conditions, the "\
-                    "relationship may be marked as viewonly=True."
-
-            raise sa_exc.ArgumentError(err)
-        else:
-            if self._user_defined_foreign_keys:
-                raise sa_exc.ArgumentError("Could not determine "
-                        "relationship direction for %s condition "
-                        "'%s', on relationship %s, using manual "
-                        "'foreign_keys' setting.  Do the columns "
-                        "in 'foreign_keys' represent all, and "
-                        "only, the 'foreign' columns in this join "
-                        "condition?  Does the %s Table already "
-                        "have adequate ForeignKey and/or "
-                        "ForeignKeyConstraint objects established "
-                        "(in which case 'foreign_keys' is usually "
-                        "unnecessary)?" 
-                        % (
-                            primary and 'primaryjoin' or 'secondaryjoin',
-                            join_condition, 
-                            self,
-                            primary and 'mapped' or 'secondary'
-                        ))
-            else:
-                raise sa_exc.ArgumentError("Could not determine "
-                        "relationship direction for %s condition "
-                        "'%s', on relationship %s. 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."
-                        % (
-                            primary and 'primaryjoin' or 'secondaryjoin', 
-                            join_condition, 
-                            self
-                        ))
-
-    def _determine_synchronize_pairs(self):
-        """Resolve 'primary'/foreign' column pairs from the primaryjoin
-        and secondaryjoin arguments.
-        
-        """
-        if self.local_remote_pairs:
-            if not self._user_defined_foreign_keys:
-                raise sa_exc.ArgumentError(
-                        "foreign_keys argument is "
-                        "required with _local_remote_pairs argument")
-            self.synchronize_pairs = []
-            for l, r in self.local_remote_pairs:
-                if r in self._user_defined_foreign_keys:
-                    self.synchronize_pairs.append((l, r))
-                elif l in self._user_defined_foreign_keys:
-                    self.synchronize_pairs.append((r, l))
-        else:
-            self.synchronize_pairs = self._sync_pairs_from_join(
-                                                self.primaryjoin, 
-                                                True)
-
-        self._calculated_foreign_keys = util.column_set(
-                                r for (l, r) in
-                                self.synchronize_pairs)
-
-        if self.secondaryjoin is not None:
-            self.secondary_synchronize_pairs = self._sync_pairs_from_join(
-                                                        self.secondaryjoin, 
-                                                        False)
-            self._calculated_foreign_keys.update(
-                                r for (l, r) in
-                                self.secondary_synchronize_pairs)
-        else:
-            self.secondary_synchronize_pairs = None
-
-    def _determine_direction(self):
-        """Determine if this relationship is one to many, many to one, 
-        many to many.
-        
-        This is derived from the primaryjoin, presence of "secondary",
-        and in the case of self-referential the "remote side".
-        
-        """
-        if self.secondaryjoin is not None:
-            self.direction = MANYTOMANY
-        elif self._refers_to_parent_table():
-
-            # self referential defaults to ONETOMANY unless the "remote"
-            # side is present and does not reference any foreign key
-            # columns
-
-            if self.local_remote_pairs:
-                remote = [r for (l, r) in self.local_remote_pairs]
-            elif self.remote_side:
-                remote = self.remote_side
-            else:
-                remote = None
-            if not remote or self._calculated_foreign_keys.difference(l for (l,
-                    r) in self.synchronize_pairs).intersection(remote):
-                self.direction = ONETOMANY
-            else:
-                self.direction = MANYTOONE
-        else:
-            parentcols = util.column_set(self.parent.mapped_table.c)
-            targetcols = util.column_set(self.mapper.mapped_table.c)
-
-            # fk collection which suggests ONETOMANY.
-            onetomany_fk = targetcols.intersection(
-                            self._calculated_foreign_keys)
+    def _setup_join_conditions(self):
+        self._join_condition = jc = relationships.JoinCondition(
+                    parent_selectable=self.parent.mapped_table,
+                    child_selectable=self.mapper.mapped_table,
+                    parent_local_selectable=self.parent.local_table,
+                    child_local_selectable=self.mapper.local_table,
+                    primaryjoin=self.primaryjoin,
+                    secondary=self.secondary,
+                    secondaryjoin=self.secondaryjoin,
+                    parent_equivalents=self.parent._equivalent_columns,
+                    child_equivalents=self.mapper._equivalent_columns,
+                    consider_as_foreign_keys=self._user_defined_foreign_keys,
+                    local_remote_pairs=self.local_remote_pairs,
+                    remote_side=self.remote_side,
+                    self_referential=self._is_self_referential,
+                    prop=self,
+                    support_sync=not self.viewonly,
+                    can_be_synced_fn=self._columns_are_mapped
+        )
+        self.primaryjoin = jc.deannotated_primaryjoin
+        self.secondaryjoin = jc.deannotated_secondaryjoin
+        self.direction = jc.direction
+        self.local_remote_pairs = jc.local_remote_pairs
+        self.remote_side = jc.remote_columns
+        self.local_columns = jc.local_columns
+        self.synchronize_pairs = jc.synchronize_pairs
+        self._calculated_foreign_keys = jc.foreign_key_columns
+        self.secondary_synchronize_pairs = jc.secondary_synchronize_pairs
 
-            # fk collection which suggests MANYTOONE.
+    def _check_conflicts(self):
+        """Test that this relationship is legal, warn about 
+        inheritance conflicts."""
 
-            manytoone_fk = parentcols.intersection(
-                            self._calculated_foreign_keys)
+        if not self.is_primary() \
+            and not mapper.class_mapper(
+                                self.parent.class_,
+                                compile=False).has_property(self.key):
+            raise sa_exc.ArgumentError("Attempting to assign a new "
+                    "relationship '%s' to a non-primary mapper on "
+                    "class '%s'.  New relationships can only be added "
+                    "to the primary mapper, i.e. the very first mapper "
+                    "created for class '%s' " % (self.key,
+                    self.parent.class_.__name__,
+                    self.parent.class_.__name__))
 
-            if onetomany_fk and manytoone_fk:
-                # fks on both sides.  do the same test only based on the
-                # local side.
-                referents = [c for (c, f) in self.synchronize_pairs]
-                onetomany_local = parentcols.intersection(referents)
-                manytoone_local = targetcols.intersection(referents)
+        # check for conflicting relationship() on superclass
+        if not self.parent.concrete:
+            for inheriting in self.parent.iterate_to_root():
+                if inheriting is not self.parent \
+                    and inheriting.has_property(self.key):
+                    util.warn("Warning: relationship '%s' on mapper "
+                              "'%s' supersedes the same relationship "
+                              "on inherited mapper '%s'; this can "
+                              "cause dependency issues during flush"
+                              % (self.key, self.parent, inheriting))
 
-                if onetomany_local and not manytoone_local:
-                    self.direction = ONETOMANY
-                elif manytoone_local and not onetomany_local:
-                    self.direction = MANYTOONE
-                else:
-                    raise sa_exc.ArgumentError(
-                            "Can't determine relationship"
-                            " direction for relationship '%s' - foreign "
-                            "key columns are present in both the parent "
-                            "and the child's mapped tables.  Specify "
-                            "'foreign_keys' argument." % self)
-            elif onetomany_fk:
-                self.direction = ONETOMANY
-            elif manytoone_fk:
-                self.direction = MANYTOONE
-            else:
-                raise sa_exc.ArgumentError("Can't determine relationship "
-                        "direction for relationship '%s' - foreign "
-                        "key columns are present in neither the parent "
-                        "nor the child's mapped tables" % self)
 
+    def _check_cascade_settings(self):
         if self.cascade.delete_orphan and not self.single_parent \
             and (self.direction is MANYTOMANY or self.direction
                  is MANYTOONE):
@@ -1300,84 +1048,24 @@ class RelationshipProperty(StrategizedProperty):
                       "relationships only."
                        % self)
 
-    def _determine_local_remote_pairs(self):
-        """Determine pairs of columns representing "local" to 
-        "remote", where "local" columns are on the parent mapper,
-        "remote" are on the target mapper.
-        
-        These pairs are used on the load side only to generate
-        lazy loading clauses.
+    def _columns_are_mapped(self, *cols):
+        """Return True if all columns in the given collection are 
+        mapped by the tables referenced by this :class:`.Relationship`.
 
         """
-        if not self.local_remote_pairs and not self.remote_side:
-            # the most common, trivial case.   Derive 
-            # local/remote pairs from the synchronize pairs.
-            eq_pairs = util.unique_list(
-                            self.synchronize_pairs + 
-                            (self.secondary_synchronize_pairs or []))
-            if self.direction is MANYTOONE:
-                self.local_remote_pairs = [(r, l) for l, r in eq_pairs]
-            else:
-                self.local_remote_pairs = eq_pairs
-
-        # "remote_side" specified, derive from the primaryjoin
-        # plus remote_side, similarly to how synchronize_pairs
-        # were determined.
-        elif self.remote_side:
-            if self.local_remote_pairs:
-                raise sa_exc.ArgumentError('remote_side argument is '
-                    'redundant against more detailed '
-                    '_local_remote_side argument.')
-            if self.direction is MANYTOONE:
-                self.local_remote_pairs = [(r, l) for (l, r) in
-                        criterion_as_pairs(self.primaryjoin,
-                        consider_as_referenced_keys=self.remote_side,
-                        any_operator=True)]
-
-            else:
-                self.local_remote_pairs = \
-                    criterion_as_pairs(self.primaryjoin,
-                        consider_as_foreign_keys=self.remote_side,
-                        any_operator=True)
-            if not self.local_remote_pairs:
-                raise sa_exc.ArgumentError('Relationship %s could '
-                        'not determine any local/remote column '
-                        'pairs from remote side argument %r'
-                        % (self, self.remote_side))
-        # else local_remote_pairs were sent explcitly via
-        # ._local_remote_pairs.
-
-        # create local_side/remote_side accessors
-        self.local_side = util.ordered_column_set(
-                            l for l, r in self.local_remote_pairs)
-        self.remote_side = util.ordered_column_set(
-                            r for l, r in self.local_remote_pairs)
-
-        # check that the non-foreign key column in the local/remote
-        # collection is mapped.  The foreign key
-        # which the individual mapped column references directly may
-        # itself be in a non-mapped table; see
-        # test.orm.test_relationships.ViewOnlyComplexJoin.test_basic
-        # for an example of this.
-        if self.direction is ONETOMANY:
-            for col in self.local_side:
-                if not self._columns_are_mapped(col):
-                    raise sa_exc.ArgumentError(
-                            "Local column '%s' is not "
-                            "part of mapping %s.  Specify remote_side "
-                            "argument to indicate which column lazy join "
-                            "condition should compare against." % (col,
-                            self.parent))
-        elif self.direction is MANYTOONE:
-            for col in self.remote_side:
-                if not self._columns_are_mapped(col):
-                    raise sa_exc.ArgumentError(
-                            "Remote column '%s' is not "
-                            "part of mapping %s. Specify remote_side "
-                            "argument to indicate which column lazy join "
-                            "condition should bind." % (col, self.mapper))
+        for c in cols:
+            if self.secondary is not None \
+                and self.secondary.c.contains_column(c):
+                continue
+            if not self.parent.mapped_table.c.contains_column(c) and \
+                not self.target.c.contains_column(c):
+                return False
+        return True
 
     def _generate_backref(self):
+        """Interpret the 'backref' instruction to create a 
+        :func:`.relationship` complementary to this one."""
+
         if not self.is_primary():
             return
         if self.backref is not None and not self.back_populates:
@@ -1391,17 +1079,27 @@ class RelationshipProperty(StrategizedProperty):
                         "'%s' on relationship '%s': property of that "
                         "name exists on mapper '%s'" % (backref_key,
                         self, mapper))
+
+            # determine primaryjoin/secondaryjoin for the
+            # backref.  Use the one we had, so that
+            # a custom join doesn't have to be specified in
+            # both directions.
             if self.secondary is not None:
-                pj = kwargs.pop('primaryjoin', self.secondaryjoin)
-                sj = kwargs.pop('secondaryjoin', self.primaryjoin)
+                # for many to many, just switch primaryjoin/
+                # secondaryjoin.   use the annotated
+                # pj/sj on the _join_condition.
+                pj = kwargs.pop('primaryjoin', self._join_condition.secondaryjoin)
+                sj = kwargs.pop('secondaryjoin', self._join_condition.primaryjoin)
             else:
-                pj = kwargs.pop('primaryjoin', self.primaryjoin)
+                pj = kwargs.pop('primaryjoin', 
+                        self._join_condition.primaryjoin_reverse_remote)
                 sj = kwargs.pop('secondaryjoin', None)
                 if sj:
                     raise sa_exc.InvalidRequestError(
-                        "Can't assign 'secondaryjoin' on a backref against "
-                        "a non-secondary relationship."
-                            )
+                        "Can't assign 'secondaryjoin' on a backref "
+                        "against a non-secondary relationship."
+                    )
+
             foreign_keys = kwargs.pop('foreign_keys',
                     self._user_defined_foreign_keys)
             parent = self.parent.primary_mapper()
@@ -1410,35 +1108,17 @@ class RelationshipProperty(StrategizedProperty):
             kwargs.setdefault('passive_updates', self.passive_updates)
             self.back_populates = backref_key
             relationship = RelationshipProperty(
-                parent,
-                self.secondary,
-                pj,
-                sj,
+                parent, self.secondary,
+                pj, sj,
                 foreign_keys=foreign_keys,
                 back_populates=self.key,
-                **kwargs
-                )
+                **kwargs)
             mapper._configure_property(backref_key, relationship)
 
         if self.back_populates:
             self._add_reverse_property(self.back_populates)
 
     def _post_init(self):
-        self.logger.info('%s setup primary join %s', self,
-                         self.primaryjoin)
-        self.logger.info('%s setup secondary join %s', self,
-                         self.secondaryjoin)
-        self.logger.info('%s synchronize pairs [%s]', self,
-                         ','.join('(%s => %s)' % (l, r) for (l, r) in
-                         self.synchronize_pairs))
-        self.logger.info('%s secondary synchronize pairs [%s]', self,
-                         ','.join('(%s => %s)' % (l, r) for (l, r) in
-                         self.secondary_synchronize_pairs or []))
-        self.logger.info('%s local/remote pairs [%s]', self,
-                         ','.join('(%s / %s)' % (l, r) for (l, r) in
-                         self.local_remote_pairs))
-        self.logger.info('%s relationship direction %s', self,
-                         self.direction)
         if self.uselist is None:
             self.uselist = self.direction is not MANYTOONE
         if not self.viewonly:
@@ -1453,20 +1133,6 @@ class RelationshipProperty(StrategizedProperty):
         strategy = self._get_strategy(strategies.LazyLoader)
         return strategy.use_get
 
-    def _refers_to_parent_table(self):
-        pt = self.parent.mapped_table
-        mt = self.mapper.mapped_table
-        for c, f in self.synchronize_pairs:
-            if (
-                pt.is_derived_from(c.table) and \
-                pt.is_derived_from(f.table) and \
-                mt.is_derived_from(c.table) and \
-                mt.is_derived_from(f.table)
-            ):
-                return True
-        else:
-            return False
-
     @util.memoized_property
     def _is_self_referential(self):
         return self.mapper.common_parent(self.parent)
@@ -1496,75 +1162,22 @@ class RelationshipProperty(StrategizedProperty):
         else:
             aliased = True
 
-        # place a barrier on the destination such that
-        # replacement traversals won't ever dig into it.
-        # its internal structure remains fixed 
-        # regardless of context.
-        dest_selectable = _shallow_annotate(
-                                dest_selectable, 
-                                {'no_replacement_traverse':True})
-
-        aliased = aliased or (source_selectable is not None)
-
-        primaryjoin, secondaryjoin, secondary = self.primaryjoin, \
-            self.secondaryjoin, self.secondary
-
-        # adjust the join condition for single table inheritance,
-        # in the case that the join is to a subclass
-        # this is analogous to the "_adjust_for_single_table_inheritance()"
-        # method in Query.
-
         dest_mapper = of_type or self.mapper
 
         single_crit = dest_mapper._single_table_criterion
-        if single_crit is not None:
-            if secondaryjoin is not None:
-                secondaryjoin = secondaryjoin & single_crit
-            else:
-                primaryjoin = primaryjoin & single_crit
-
-        if aliased:
-            if secondary is not None:
-                secondary = secondary.alias()
-                primary_aliasizer = ClauseAdapter(secondary)
-                secondary_aliasizer = \
-                    ClauseAdapter(dest_selectable,
-                        equivalents=self.mapper._equivalent_columns).\
-                        chain(primary_aliasizer)
-                if source_selectable is not None:
-                    primary_aliasizer = \
-                        ClauseAdapter(secondary).\
-                            chain(ClauseAdapter(source_selectable,
-                            equivalents=self.parent._equivalent_columns))
-                secondaryjoin = \
-                    secondary_aliasizer.traverse(secondaryjoin)
-            else:
-                primary_aliasizer = ClauseAdapter(dest_selectable,
-                        exclude=self.local_side,
-                        equivalents=self.mapper._equivalent_columns)
-                if source_selectable is not None:
-                    primary_aliasizer.chain(
-                        ClauseAdapter(source_selectable,
-                            exclude=self.remote_side,
-                            equivalents=self.parent._equivalent_columns))
-                secondary_aliasizer = None
-            primaryjoin = primary_aliasizer.traverse(primaryjoin)
-            target_adapter = secondary_aliasizer or primary_aliasizer
-            target_adapter.include = target_adapter.exclude = None
-        else:
-            target_adapter = None
+        aliased = aliased or (source_selectable is not None)
+
+        primaryjoin, secondaryjoin, secondary, target_adapter, dest_selectable = \
+            self._join_condition.join_targets(
+                source_selectable, dest_selectable, aliased, single_crit
+            )
         if source_selectable is None:
             source_selectable = self.parent.local_table
         if dest_selectable is None:
             dest_selectable = self.mapper.local_table
-        return (
-            primaryjoin,
-            secondaryjoin,
-            source_selectable,
-            dest_selectable,
-            secondary,
-            target_adapter,
-            )
+        return (primaryjoin, secondaryjoin, source_selectable,
+            dest_selectable, secondary, target_adapter)
+
 
 PropertyLoader = RelationProperty = RelationshipProperty
 log.class_logger(RelationshipProperty)
diff --git a/lib/sqlalchemy/orm/relationships.py b/lib/sqlalchemy/orm/relationships.py
new file mode 100644
index 000000000..4c64e855f
--- /dev/null
+++ b/lib/sqlalchemy/orm/relationships.py
@@ -0,0 +1,856 @@
+# orm/relationships.py
+# Copyright (C) 2005-2012 the SQLAlchemy authors and contributors <see AUTHORS file>
+#
+# This module is part of SQLAlchemy and is released under
+# the MIT License: http://www.opensource.org/licenses/mit-license.php
+
+"""Heuristics related to join conditions as used in 
+:func:`.relationship`.
+
+Provides the :class:`.JoinCondition` object, which encapsulates
+SQL annotation and aliasing behavior focused on the `primaryjoin`
+and `secondaryjoin` aspects of :func:`.relationship`.
+
+"""
+
+from sqlalchemy import sql, util, log, exc as sa_exc, schema
+from sqlalchemy.sql.util import ClauseAdapter, criterion_as_pairs, \
+    join_condition, _shallow_annotate, visit_binary_product,\
+    _deep_deannotate
+from sqlalchemy.sql import operators, expression, visitors
+from sqlalchemy.orm.interfaces import MANYTOMANY, MANYTOONE, ONETOMANY
+
+def remote(expr):
+    return _annotate_columns(expr, {"remote":True})
+
+def foreign(expr):
+    return _annotate_columns(expr, {"foreign":True})
+
+def remote_foreign(expr):
+    return _annotate_columns(expr, {"foreign":True, 
+                                "remote":True})
+
+def _annotate_columns(element, annotations):
+    def clone(elem):
+        if isinstance(elem, expression.ColumnClause):
+            elem = elem._annotate(annotations.copy())
+        elem._copy_internals(clone=clone)
+        return elem
+
+    if element is not None:
+        element = clone(element)
+    return element
+
+class JoinCondition(object):
+    def __init__(self, 
+                    parent_selectable, 
+                    child_selectable,
+                    parent_local_selectable,
+                    child_local_selectable,
+                    primaryjoin=None,
+                    secondary=None,
+                    secondaryjoin=None,
+                    parent_equivalents=None,
+                    child_equivalents=None,
+                    consider_as_foreign_keys=None,
+                    local_remote_pairs=None,
+                    remote_side=None,
+                    self_referential=False,
+                    prop=None,
+                    support_sync=True,
+                    can_be_synced_fn=lambda *c: True
+                    ):
+        self.parent_selectable = parent_selectable
+        self.parent_local_selectable = parent_local_selectable
+        self.child_selectable = child_selectable
+        self.child_local_selectable = child_local_selectable
+        self.parent_equivalents = parent_equivalents
+        self.child_equivalents = child_equivalents
+        self.primaryjoin = primaryjoin
+        self.secondaryjoin = secondaryjoin
+        self.secondary = secondary
+        self.consider_as_foreign_keys = consider_as_foreign_keys
+        self._local_remote_pairs = local_remote_pairs
+        self._remote_side = remote_side
+        self.prop = prop
+        self.self_referential = self_referential
+        self.support_sync = support_sync
+        self.can_be_synced_fn = can_be_synced_fn
+        self._determine_joins()
+        self._annotate_fks()
+        self._annotate_remote()
+        self._annotate_local()
+        self._setup_pairs()
+        self._check_foreign_cols(self.primaryjoin, True)
+        if self.secondaryjoin is not None:
+            self._check_foreign_cols(self.secondaryjoin, False)
+        self._determine_direction()
+        self._check_remote_side()
+        self._log_joins()
+
+    def _log_joins(self):
+        if self.prop is None:
+            return
+        log = self.prop.logger
+        log.info('%s setup primary join %s', self,
+                         self.primaryjoin)
+        log.info('%s setup secondary join %s', self,
+                         self.secondaryjoin)
+        log.info('%s synchronize pairs [%s]', self,
+                         ','.join('(%s => %s)' % (l, r) for (l, r) in
+                         self.synchronize_pairs))
+        log.info('%s secondary synchronize pairs [%s]', self,
+                         ','.join('(%s => %s)' % (l, r) for (l, r) in
+                         self.secondary_synchronize_pairs or []))
+        log.info('%s local/remote pairs [%s]', self,
+                         ','.join('(%s / %s)' % (l, r) for (l, r) in
+                         self.local_remote_pairs))
+        log.info('%s relationship direction %s', self,
+                         self.direction)
+
+    def _determine_joins(self):
+        """Determine the 'primaryjoin' and 'secondaryjoin' attributes,
+        if not passed to the constructor already.
+
+        This is based on analysis of the foreign key relationships
+        between the parent and target mapped selectables.
+
+        """
+        if self.secondaryjoin is not None and self.secondary is None:
+            raise sa_exc.ArgumentError(
+                    "Property %s specified with secondary "
+                    "join condition but "
+                    "no secondary argument" % self.prop)
+
+        # find a join between the given mapper's mapped table and
+        # the given table. will try the mapper's local table first
+        # for more specificity, then if not found will try the more
+        # general mapped table, which in the case of inheritance is
+        # a join.
+        try:
+            if self.secondary is not None:
+                if self.secondaryjoin is None:
+                    self.secondaryjoin = \
+                        join_condition(
+                                self.child_selectable, 
+                                self.secondary,
+                                a_subset=self.child_local_selectable,
+                                consider_as_foreign_keys=\
+                                    self.consider_as_foreign_keys or None
+                                )
+                if self.primaryjoin is None:
+                    self.primaryjoin = \
+                        join_condition(
+                                self.parent_selectable, 
+                                self.secondary, 
+                                a_subset=self.parent_local_selectable,
+                                consider_as_foreign_keys=\
+                                    self.consider_as_foreign_keys or None
+                                )
+            else:
+                if self.primaryjoin is None:
+                    self.primaryjoin = \
+                        join_condition(
+                                self.parent_selectable, 
+                                self.child_selectable, 
+                                a_subset=self.parent_local_selectable,
+                                consider_as_foreign_keys=\
+                                    self.consider_as_foreign_keys or None
+                                )
+        except sa_exc.NoForeignKeysError, nfke:
+            if self.secondary is not None:
+                raise sa_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."
+                        % (self.prop, self.secondary))
+            else:
+                raise sa_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."
+                        % self.prop)
+        except sa_exc.AmbiguousForeignKeysError, afke:
+            if self.secondary is not None:
+                raise sa_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."
+                        % (self.prop, self.secondary))
+            else:
+                raise sa_exc.AmbiguousForeignKeysError(
+                        "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."
+                        % self.prop)
+
+    @util.memoized_property
+    def primaryjoin_reverse_remote(self):
+        """Return the primaryjoin condition suitable for the 
+        "reverse" direction.  
+        
+        If the primaryjoin was delivered here with pre-existing
+        "remote" annotations, the local/remote annotations
+        are reversed.  Otherwise, the local/remote annotations
+        are removed.
+        
+        """
+        if self._has_remote_annotations:
+            def replace(element):
+                if "remote" in element._annotations:
+                    v = element._annotations.copy()
+                    del v['remote']
+                    v['local'] = True
+                    return element._with_annotations(v)
+                elif "local" in element._annotations:
+                    v = element._annotations.copy()
+                    del v['local']
+                    v['remote'] = True
+                    return element._with_annotations(v)
+            return visitors.replacement_traverse(
+                                self.primaryjoin, {}, replace)
+        else:
+            if self._has_foreign_annotations:
+                # TODO: coverage
+                return _deep_deannotate(self.primaryjoin, 
+                                values=("local", "remote"))
+            else:
+                return _deep_deannotate(self.primaryjoin)
+
+    def _has_annotation(self, clause, annotation):
+        for col in visitors.iterate(clause, {}):
+            if annotation in col._annotations:
+                return True
+        else:
+            return False
+
+    @util.memoized_property
+    def _has_foreign_annotations(self):
+        return self._has_annotation(self.primaryjoin, "foreign")
+
+    @util.memoized_property
+    def _has_remote_annotations(self):
+        return self._has_annotation(self.primaryjoin, "remote")
+
+    def _annotate_fks(self):
+        """Annotate the primaryjoin and secondaryjoin
+        structures with 'foreign' annotations marking columns
+        considered as foreign.
+        
+        """
+        if self._has_foreign_annotations:
+            return
+
+        if self.consider_as_foreign_keys:
+            self._annotate_from_fk_list()
+        else:
+            self._annotate_present_fks()
+
+    def _annotate_from_fk_list(self):
+        def check_fk(col):
+            if col in self.consider_as_foreign_keys:
+                return col._annotate({"foreign":True})
+        self.primaryjoin = visitors.replacement_traverse(
+            self.primaryjoin,
+            {},
+            check_fk
+        )
+        if self.secondaryjoin is not None:
+            self.secondaryjoin = visitors.replacement_traverse(
+                self.secondaryjoin,
+                {},
+                check_fk
+            )
+
+    def _annotate_present_fks(self):
+        if self.secondary is not None:
+            secondarycols = util.column_set(self.secondary.c)
+        else:
+            secondarycols = set()
+        def is_foreign(a, b):
+            if isinstance(a, schema.Column) and \
+                        isinstance(b, schema.Column):
+                if a.references(b):
+                    return a
+                elif b.references(a):
+                    return b
+
+            if secondarycols:
+                if a in secondarycols and b not in secondarycols:
+                    return a
+                elif b in secondarycols and a not in secondarycols:
+                    return b
+
+        def visit_binary(binary):
+            if not isinstance(binary.left, sql.ColumnElement) or \
+                        not isinstance(binary.right, sql.ColumnElement):
+                return
+
+            if "foreign" not in binary.left._annotations and \
+                "foreign" not in binary.right._annotations:
+                col = is_foreign(binary.left, binary.right)
+                if col is not None:
+                    if col.compare(binary.left):
+                        binary.left = binary.left._annotate(
+                                            {"foreign":True})
+                    elif col.compare(binary.right):
+                        binary.right = binary.right._annotate(
+                                            {"foreign":True})
+
+        self.primaryjoin = visitors.cloned_traverse(
+            self.primaryjoin,
+            {},
+            {"binary":visit_binary}
+        )
+        if self.secondaryjoin is not None:
+            self.secondaryjoin = visitors.cloned_traverse(
+                self.secondaryjoin,
+                {},
+                {"binary":visit_binary}
+            )
+
+    def _refers_to_parent_table(self):
+        """Return True if the join condition contains column
+        comparisons where both columns are in both tables.
+        
+        """
+        pt = self.parent_selectable
+        mt = self.child_selectable
+        result = [False]
+        def visit_binary(binary):
+            c, f = binary.left, binary.right
+            if (
+                isinstance(c, expression.ColumnClause) and \
+                isinstance(f, expression.ColumnClause) and \
+                pt.is_derived_from(c.table) and \
+                pt.is_derived_from(f.table) and \
+                mt.is_derived_from(c.table) and \
+                mt.is_derived_from(f.table)
+            ):
+                result[0] = True
+        visitors.traverse(
+                    self.primaryjoin,
+                    {},
+                    {"binary":visit_binary}
+                )
+        return result[0]
+
+    def _tables_overlap(self):
+        """Return True if parent/child tables have some overlap."""
+
+        return self.parent_selectable.is_derived_from(
+                        self.child_local_selectable) or \
+            self.child_selectable.is_derived_from(
+                        self.parent_local_selectable)
+
+    def _annotate_remote(self):
+        """Annotate the primaryjoin and secondaryjoin
+        structures with 'remote' annotations marking columns
+        considered as part of the 'remote' side.
+        
+        """
+        if self._has_remote_annotations:
+            return
+
+        parentcols = util.column_set(self.parent_selectable.c)
+
+        if self.secondary is not None:
+            self._annotate_remote_secondary()
+        elif self._local_remote_pairs or self._remote_side:
+            self._annotate_remote_from_args()
+        elif self._refers_to_parent_table():
+            self._annotate_selfref(lambda col:"foreign" in col._annotations)
+        elif self._tables_overlap():
+            self._annotate_remote_with_overlap()
+        else:
+            self._annotate_remote_distinct_selectables()
+
+    def _annotate_remote_secondary(self):
+        """annotate 'remote' in primaryjoin, secondaryjoin
+        when 'secondary' is present.
+        
+        """
+        def repl(element):
+            if self.secondary.c.contains_column(element):
+                return element._annotate({"remote":True})
+        self.primaryjoin = visitors.replacement_traverse(
+                                    self.primaryjoin, {},  repl)
+        self.secondaryjoin = visitors.replacement_traverse(
+                                    self.secondaryjoin, {}, repl)
+
+    def _annotate_selfref(self, fn):
+        """annotate 'remote' in primaryjoin, secondaryjoin
+        when the relationship is detected as self-referential.
+        
+        """
+        def visit_binary(binary):
+            equated = binary.left.compare(binary.right)
+            if isinstance(binary.left, expression.ColumnClause) and \
+                isinstance(binary.right, expression.ColumnClause):
+                # assume one to many - FKs are "remote"
+                if fn(binary.left):
+                    binary.left = binary.left._annotate({"remote":True})
+                if fn(binary.right) and \
+                    not equated:
+                    binary.right = binary.right._annotate(
+                                        {"remote":True})
+            else:
+                self._warn_non_column_elements()
+
+        self.primaryjoin = visitors.cloned_traverse(
+                                self.primaryjoin, {}, 
+                                {"binary":visit_binary})
+
+    def _annotate_remote_from_args(self):
+        """annotate 'remote' in primaryjoin, secondaryjoin
+        when the 'remote_side' or '_local_remote_pairs'
+        arguments are used.
+        
+        """
+        if self._local_remote_pairs:
+            if self._remote_side:
+                raise sa_exc.ArgumentError(
+                        "remote_side argument is redundant "
+                        "against more detailed _local_remote_side "
+                        "argument.")
+
+            remote_side = [r for (l, r) in self._local_remote_pairs]
+        else:
+            remote_side = self._remote_side
+
+        if self._refers_to_parent_table():
+            self._annotate_selfref(lambda col:col in remote_side)
+        else:
+            def repl(element):
+                if element in remote_side:
+                    return element._annotate({"remote":True})
+            self.primaryjoin = visitors.replacement_traverse(
+                                        self.primaryjoin, {},  repl)
+
+    def _annotate_remote_with_overlap(self):
+        """annotate 'remote' in primaryjoin, secondaryjoin
+        when the parent/child tables have some set of 
+        tables in common, though is not a fully self-referential
+        relationship.
+        
+        """
+        def visit_binary(binary):
+            binary.left, binary.right = proc_left_right(binary.left, 
+                                            binary.right)
+            binary.right, binary.left = proc_left_right(binary.right, 
+                                            binary.left)
+        def proc_left_right(left, right):
+            if isinstance(left, expression.ColumnClause) and \
+                isinstance(right, expression.ColumnClause):
+                if self.child_selectable.c.contains_column(right) and \
+                    self.parent_selectable.c.contains_column(left):
+                    right = right._annotate({"remote":True})
+            else:
+               self._warn_non_column_elements()
+
+            return left, right
+
+        self.primaryjoin = visitors.cloned_traverse(
+                                    self.primaryjoin, {}, 
+                                    {"binary":visit_binary})
+
+    def _annotate_remote_distinct_selectables(self):
+        """annotate 'remote' in primaryjoin, secondaryjoin
+        when the parent/child tables are entirely 
+        separate.
+        
+        """
+        def repl(element):
+            if self.child_selectable.c.contains_column(element) and \
+                (
+                    not self.parent_local_selectable.c.\
+                            contains_column(element)
+                    or self.child_local_selectable.c.\
+                            contains_column(element)
+                ):
+                return element._annotate({"remote":True})
+        self.primaryjoin = visitors.replacement_traverse(
+                                    self.primaryjoin, {},  repl)
+
+    def _warn_non_column_elements(self):
+        util.warn(
+            "Non-simple column elements in primary "
+            "join condition for property %s - consider using "
+            "remote() annotations to mark the remote side."
+            % self.prop
+        )
+
+    def _annotate_local(self):
+        """Annotate the primaryjoin and secondaryjoin 
+        structures with 'local' annotations.
+        
+        This annotates all column elements found 
+        simultaneously in the parent table 
+        and the join condition that don't have a 
+        'remote' annotation set up from 
+        _annotate_remote() or user-defined.
+        
+        """
+        if self._has_annotation(self.primaryjoin, "local"):
+            return
+
+        parentcols = util.column_set(self.parent_selectable.c)
+
+        if self._local_remote_pairs:
+            local_side = util.column_set([l for (l, r) 
+                                in self._local_remote_pairs])
+        else:
+            local_side = util.column_set(self.parent_selectable.c)
+
+        def locals_(elem):
+            if "remote" not in elem._annotations and \
+                elem in local_side:
+                return elem._annotate({"local":True})
+        self.primaryjoin = visitors.replacement_traverse(
+                self.primaryjoin, {}, locals_
+            )
+
+    def _check_remote_side(self):
+        if not self.local_remote_pairs:
+            raise sa_exc.ArgumentError('Relationship %s could '
+                    'not determine any unambiguous local/remote column '
+                    'pairs based on join condition and remote_side '
+                    'arguments.  '
+                    'Consider using the remote() annotation to '
+                    'accurately mark those elements of the join '
+                    'condition that are on the remote side of '
+                    'the relationship.'
+                    % (self.prop, ))
+
+    def _check_foreign_cols(self, join_condition, primary):
+        """Check the foreign key columns collected and emit error 
+        messages."""
+
+        can_sync = False
+
+        foreign_cols = self._gather_columns_with_annotation(
+                                join_condition, "foreign")
+
+        has_foreign = bool(foreign_cols)
+
+        if primary:
+            can_sync = bool(self.synchronize_pairs)
+        else:
+            can_sync = bool(self.secondary_synchronize_pairs)
+
+        if self.support_sync and can_sync or \
+            (not self.support_sync and has_foreign):
+            return
+
+        # from here below is just determining the best error message
+        # to report.  Check for a join condition using any operator 
+        # (not just ==), perhaps they need to turn on "viewonly=True".
+        if self.support_sync and has_foreign and not can_sync:
+            err = "Could not locate any simple equality expressions "\
+                    "involving locally mapped foreign key columns for "\
+                    "%s join condition "\
+                    "'%s' on relationship %s." % (
+                        primary and 'primary' or 'secondary', 
+                        join_condition, 
+                        self.prop
+                    )
+            err += \
+                "  Ensure that referencing columns are associated "\
+                "with a ForeignKey or ForeignKeyConstraint, or are "\
+                "annotated in the join condition with the foreign() "\
+                "annotation. To allow comparison operators other than "\
+                "'==', the relationship can be marked as viewonly=True."
+
+            raise sa_exc.ArgumentError(err)
+        else:
+            err = "Could not locate any relevant foreign key columns "\
+                    "for %s join condition '%s' on relationship %s." % (
+                        primary and 'primary' or 'secondary', 
+                        join_condition, 
+                        self.prop
+                    )
+            err += \
+                '  Ensure that referencing columns are associated '\
+                'with a ForeignKey or ForeignKeyConstraint, or are '\
+                'annotated in the join condition with the foreign() '\
+                'annotation.'
+            raise sa_exc.ArgumentError(err)
+
+    def _determine_direction(self):
+        """Determine if this relationship is one to many, many to one, 
+        many to many.
+
+        """
+        if self.secondaryjoin is not None:
+            self.direction = MANYTOMANY
+        else:
+            parentcols = util.column_set(self.parent_selectable.c)
+            targetcols = util.column_set(self.child_selectable.c)
+
+            # fk collection which suggests ONETOMANY.
+            onetomany_fk = targetcols.intersection(
+                            self.foreign_key_columns)
+
+            # fk collection which suggests MANYTOONE.
+
+            manytoone_fk = parentcols.intersection(
+                            self.foreign_key_columns)
+
+            if onetomany_fk and manytoone_fk:
+                # fks on both sides.  test for overlap of local/remote
+                # with foreign key
+                self_equated = self.remote_columns.intersection(
+                                        self.local_columns
+                                    )
+                onetomany_local = self.remote_columns.\
+                                    intersection(self.foreign_key_columns).\
+                                    difference(self_equated)
+                manytoone_local = self.local_columns.\
+                                    intersection(self.foreign_key_columns).\
+                                    difference(self_equated)
+                if onetomany_local and not manytoone_local:
+                    self.direction = ONETOMANY
+                elif manytoone_local and not onetomany_local:
+                    self.direction = MANYTOONE
+                else:
+                    raise 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 are marked as foreign, "
+                        "either via the foreign() annotation or "
+                        "via the foreign_keys argument."
+                         % self.prop)
+            elif onetomany_fk:
+                self.direction = ONETOMANY
+            elif manytoone_fk:
+                self.direction = MANYTOONE
+            else:
+                raise sa_exc.ArgumentError("Can't determine relationship "
+                        "direction for relationship '%s' - foreign "
+                        "key columns are present in neither the parent "
+                        "nor the child's mapped tables" % self.prop)
+
+    def _deannotate_pairs(self, collection):
+        """provide deannotation for the various lists of 
+        pairs, so that using them in hashes doesn't incur
+        high-overhead __eq__() comparisons against
+        original columns mapped.
+        
+        """
+        return [(x._deannotate(), y._deannotate()) 
+            for x, y in collection]
+
+    def _setup_pairs(self):
+        sync_pairs = []
+        lrp = util.OrderedSet([])
+        secondary_sync_pairs = []
+
+        def go(joincond, collection):
+            def visit_binary(binary, left, right):
+                if "remote" in right._annotations and \
+                    "remote" not in left._annotations and \
+                    self.can_be_synced_fn(left):
+                    lrp.add((left, right))
+                elif "remote" in left._annotations and \
+                    "remote" not in right._annotations and \
+                    self.can_be_synced_fn(right):
+                    lrp.add((right, left))
+                if binary.operator is operators.eq and \
+                    self.can_be_synced_fn(left, right):
+                    if "foreign" in right._annotations:
+                        collection.append((left, right))
+                    elif "foreign" in left._annotations:
+                        collection.append((right, left))
+            visit_binary_product(visit_binary, joincond)
+
+        for joincond, collection in [
+            (self.primaryjoin, sync_pairs),
+            (self.secondaryjoin, secondary_sync_pairs)
+        ]:
+            if joincond is None:
+                continue
+            go(joincond, collection)
+
+        self.local_remote_pairs = self._deannotate_pairs(lrp)
+        self.synchronize_pairs = self._deannotate_pairs(sync_pairs)
+        self.secondary_synchronize_pairs = self._deannotate_pairs(secondary_sync_pairs)
+
+    @util.memoized_property
+    def remote_columns(self):
+        return self._gather_join_annotations("remote")
+
+    @util.memoized_property
+    def local_columns(self):
+        return self._gather_join_annotations("local")
+
+    @util.memoized_property
+    def foreign_key_columns(self):
+        return self._gather_join_annotations("foreign")
+
+    @util.memoized_property
+    def deannotated_primaryjoin(self):
+        return _deep_deannotate(self.primaryjoin)
+
+    @util.memoized_property
+    def deannotated_secondaryjoin(self):
+        if self.secondaryjoin is not None:
+            return _deep_deannotate(self.secondaryjoin)
+        else:
+            return None
+
+    def _gather_join_annotations(self, annotation):
+        s = set(
+            self._gather_columns_with_annotation(
+                        self.primaryjoin, annotation)
+        )
+        if self.secondaryjoin is not None:
+            s.update(
+                self._gather_columns_with_annotation(
+                        self.secondaryjoin, annotation)
+            )
+        return set([x._deannotate() for x in s])
+
+    def _gather_columns_with_annotation(self, clause, *annotation):
+        annotation = set(annotation)
+        return set([
+            col for col in visitors.iterate(clause, {})
+            if annotation.issubset(col._annotations)
+        ])
+
+
+    def join_targets(self, source_selectable, 
+                            dest_selectable,
+                            aliased,
+                            single_crit=None):
+        """Given a source and destination selectable, create a
+        join between them.
+
+        This takes into account aliasing the join clause
+        to reference the appropriate corresponding columns
+        in the target objects, as well as the extra child
+        criterion, equivalent column sets, etc.
+
+        """
+
+        # place a barrier on the destination such that
+        # replacement traversals won't ever dig into it.
+        # its internal structure remains fixed 
+        # regardless of context.
+        dest_selectable = _shallow_annotate(
+                                dest_selectable, 
+                                {'no_replacement_traverse':True})
+
+        primaryjoin, secondaryjoin, secondary = self.primaryjoin, \
+            self.secondaryjoin, self.secondary
+
+        # adjust the join condition for single table inheritance,
+        # in the case that the join is to a subclass
+        # this is analogous to the 
+        # "_adjust_for_single_table_inheritance()" method in Query.
+
+        if single_crit is not None:
+            if secondaryjoin is not None:
+                secondaryjoin = secondaryjoin & single_crit
+            else:
+                primaryjoin = primaryjoin & single_crit
+
+        if aliased:
+            if secondary is not None:
+                secondary = secondary.alias()
+                primary_aliasizer = ClauseAdapter(secondary)
+                secondary_aliasizer = \
+                    ClauseAdapter(dest_selectable,
+                        equivalents=self.child_equivalents).\
+                        chain(primary_aliasizer)
+                if source_selectable is not None:
+                    primary_aliasizer = \
+                        ClauseAdapter(secondary).\
+                            chain(ClauseAdapter(source_selectable,
+                            equivalents=self.parent_equivalents))
+                secondaryjoin = \
+                    secondary_aliasizer.traverse(secondaryjoin)
+            else:
+                primary_aliasizer = ClauseAdapter(dest_selectable,
+                        exclude_fn=lambda c: "local" in c._annotations,
+                        equivalents=self.child_equivalents)
+                if source_selectable is not None:
+                    primary_aliasizer.chain(
+                        ClauseAdapter(source_selectable,
+                            exclude_fn=lambda c: "remote" in c._annotations,
+                            equivalents=self.parent_equivalents))
+                secondary_aliasizer = None
+
+            primaryjoin = primary_aliasizer.traverse(primaryjoin)
+            target_adapter = secondary_aliasizer or primary_aliasizer
+            target_adapter.exclude_fn = None
+        else:
+            target_adapter = None
+        return primaryjoin, secondaryjoin, secondary, \
+                        target_adapter, dest_selectable
+
+    def create_lazy_clause(self, reverse_direction=False):
+        binds = util.column_dict()
+        lookup = util.column_dict()
+        equated_columns = util.column_dict()
+
+        if reverse_direction and self.secondaryjoin is None:
+            for l, r in self.local_remote_pairs:
+                _list = lookup.setdefault(r, [])
+                _list.append((r, l))
+                equated_columns[l] = r
+        else:
+            for l, r in self.local_remote_pairs:
+                _list = lookup.setdefault(l, [])
+                _list.append((l, r))
+                equated_columns[r] = l
+
+        def col_to_bind(col):
+            if col in lookup:
+                for tobind, equated in lookup[col]:
+                    if equated in binds:
+                        return None
+                if col not in binds:
+                    binds[col] = sql.bindparam(
+                        None, None, type_=col.type, unique=True)
+                return binds[col]
+            return None
+
+        lazywhere = self.deannotated_primaryjoin
+
+        if self.deannotated_secondaryjoin is None or not reverse_direction:
+            lazywhere = visitors.replacement_traverse(
+                                            lazywhere, {}, col_to_bind) 
+
+        if self.deannotated_secondaryjoin is not None:
+            secondaryjoin = self.deannotated_secondaryjoin
+            if reverse_direction:
+                secondaryjoin = visitors.replacement_traverse(
+                                            secondaryjoin, {}, col_to_bind)
+            lazywhere = sql.and_(lazywhere, secondaryjoin)
+
+        bind_to_col = dict((binds[col].key, col) for col in binds)
+
+        return lazywhere, bind_to_col, equated_columns
+
+
+
diff --git a/lib/sqlalchemy/orm/strategies.py b/lib/sqlalchemy/orm/strategies.py
index 37980e111..2e09ccbbe 100644
--- a/lib/sqlalchemy/orm/strategies.py
+++ b/lib/sqlalchemy/orm/strategies.py
@@ -324,14 +324,14 @@ class LazyLoader(AbstractRelationshipLoader):
 
     def init(self):
         super(LazyLoader, self).init()
+        join_condition = self.parent_property._join_condition
         self._lazywhere, \
         self._bind_to_col, \
-        self._equated_columns = self._create_lazy_clause(self.parent_property)
+        self._equated_columns = join_condition.create_lazy_clause()
 
         self._rev_lazywhere, \
         self._rev_bind_to_col, \
-        self._rev_equated_columns = self._create_lazy_clause(
-                                                self.parent_property, 
+        self._rev_equated_columns = join_condition.create_lazy_clause(
                                                 reverse_direction=True)
 
         self.logger.info("%s lazy loading clause %s", self, self._lazywhere)
@@ -617,49 +617,6 @@ class LazyLoader(AbstractRelationshipLoader):
 
             return reset_for_lazy_callable, None, None
 
-    @classmethod
-    def _create_lazy_clause(cls, prop, reverse_direction=False):
-        binds = util.column_dict()
-        lookup = util.column_dict()
-        equated_columns = util.column_dict()
-
-        if reverse_direction and prop.secondaryjoin is None:
-            for l, r in prop.local_remote_pairs:
-                _list = lookup.setdefault(r, [])
-                _list.append((r, l))
-                equated_columns[l] = r
-        else:
-            for l, r in prop.local_remote_pairs:
-                _list = lookup.setdefault(l, [])
-                _list.append((l, r))
-                equated_columns[r] = l
-
-        def col_to_bind(col):
-            if col in lookup:
-                for tobind, equated in lookup[col]:
-                    if equated in binds:
-                        return None
-                if col not in binds:
-                    binds[col] = sql.bindparam(None, None, type_=col.type, unique=True)
-                return binds[col]
-            return None
-
-        lazywhere = prop.primaryjoin
-
-        if prop.secondaryjoin is None or not reverse_direction:
-            lazywhere = visitors.replacement_traverse(
-                                            lazywhere, {}, col_to_bind) 
-
-        if prop.secondaryjoin is not None:
-            secondaryjoin = prop.secondaryjoin
-            if reverse_direction:
-                secondaryjoin = visitors.replacement_traverse(
-                                            secondaryjoin, {}, col_to_bind)
-            lazywhere = sql.and_(lazywhere, secondaryjoin)
-
-        bind_to_col = dict((binds[col].key, col) for col in binds)
-
-        return lazywhere, bind_to_col, equated_columns
 
 log.class_logger(LazyLoader)
 
@@ -785,7 +742,8 @@ class SubqueryLoader(AbstractRelationshipLoader):
             leftmost_mapper, leftmost_prop = \
                                     subq_mapper, \
                                     subq_mapper._props[subq_path[1]]
-        leftmost_cols, remote_cols = self._local_remote_columns(leftmost_prop)
+
+        leftmost_cols = leftmost_prop.local_columns
 
         leftmost_attr = [
             leftmost_mapper._columntoproperty[c].class_attribute
@@ -846,8 +804,7 @@ class SubqueryLoader(AbstractRelationshipLoader):
             # self.parent is more specific than subq_path[-2]
             parent_alias = mapperutil.AliasedClass(self.parent)
 
-        local_cols, remote_cols = \
-                        self._local_remote_columns(self.parent_property)
+        local_cols = self.parent_property.local_columns
 
         local_attr = [
             getattr(parent_alias, self.parent._columntoproperty[c].key)
@@ -881,17 +838,6 @@ class SubqueryLoader(AbstractRelationshipLoader):
                 q = q.join(attr, aliased=middle, from_joinpoint=True)
         return q
 
-    def _local_remote_columns(self, prop):
-        if prop.secondary is None:
-            return zip(*prop.local_remote_pairs)
-        else:
-            return \
-                [p[0] for p in prop.synchronize_pairs],\
-                [
-                    p[0] for p in prop.
-                                        secondary_synchronize_pairs
-                ]
-
     def _setup_options(self, q, subq_path, orig_query):
         # propagate loader options etc. to the new query.
         # these will fire relative to subq_path.
@@ -930,7 +876,7 @@ class SubqueryLoader(AbstractRelationshipLoader):
         if ('subquery', reduced_path) not in context.attributes:
             return None, None, None
 
-        local_cols, remote_cols = self._local_remote_columns(self.parent_property)
+        local_cols = self.parent_property.local_columns
 
         q = context.attributes[('subquery', reduced_path)]
 
diff --git a/lib/sqlalchemy/orm/util.py b/lib/sqlalchemy/orm/util.py
index 197c0c4c1..c14c22bac 100644
--- a/lib/sqlalchemy/orm/util.py
+++ b/lib/sqlalchemy/orm/util.py
@@ -393,7 +393,21 @@ def _orm_annotate(element, exclude=None):
     """
     return sql_util._deep_annotate(element, {'_orm_adapt':True}, exclude)
 
-_orm_deannotate = sql_util._deep_deannotate
+def _orm_deannotate(element):
+    """Remove annotations that link a column to a particular mapping.
+    
+    Note this doesn't affect "remote" and "foreign" annotations
+    passed by the :func:`.orm.foreign` and :func:`.orm.remote`
+    annotators.
+    
+    """
+
+    return sql_util._deep_deannotate(element, 
+                values=("_orm_adapt", "parententity")
+            )
+
+def _orm_full_deannotate(element):
+    return sql_util._deep_deannotate(element)
 
 class _ORMJoin(expression.Join):
     """Extend Join to support ORM constructs as input."""
diff --git a/lib/sqlalchemy/sql/expression.py b/lib/sqlalchemy/sql/expression.py
index f9a3863da..d8ad7c3fa 100644
--- a/lib/sqlalchemy/sql/expression.py
+++ b/lib/sqlalchemy/sql/expression.py
@@ -1584,18 +1584,35 @@ class ClauseElement(Visitable):
             return id(self)
 
     def _annotate(self, values):
-        """return a copy of this ClauseElement with the given annotations
-        dictionary.
+        """return a copy of this ClauseElement with annotations
+        updated by the given dictionary.
 
         """
         return sqlutil.Annotated(self, values)
 
-    def _deannotate(self):
-        """return a copy of this ClauseElement with an empty annotations
-        dictionary.
+    def _with_annotations(self, values):
+        """return a copy of this ClauseElement with annotations
+        replaced by the given dictionary.
 
         """
-        return self._clone()
+        return sqlutil.Annotated(self, values)
+
+    def _deannotate(self, values=None, clone=False):
+        """return a copy of this :class:`.ClauseElement` with annotations
+        removed.
+        
+        :param values: optional tuple of individual values
+         to remove.
+
+        """
+        if clone:
+            # clone is used when we are also copying
+            # the expression for a deep deannotation
+            return self._clone()
+        else:
+            # if no clone, since we have no annotations we return
+            # self
+            return self
 
     def unique_params(self, *optionaldict, **kwargs):
         """Return a copy with :func:`bindparam()` elements replaced.
@@ -2195,7 +2212,7 @@ class ColumnElement(ClauseElement, _CompareMixin):
         for oth in to_compare:
             if use_proxies and self.shares_lineage(oth):
                 return True
-            elif oth is self:
+            elif hash(oth) == hash(self):
                 return True
         else:
             return False
@@ -3403,6 +3420,10 @@ class _BinaryExpression(ColumnElement):
             raise TypeError("Boolean value of this clause is not defined")
 
     @property
+    def is_comparison(self):
+        return operators.is_comparison(self.operator)
+
+    @property
     def _from_objects(self):
         return self.left._from_objects + self.right._from_objects
 
diff --git a/lib/sqlalchemy/sql/operators.py b/lib/sqlalchemy/sql/operators.py
index 89f0aaee1..b86b50db4 100644
--- a/lib/sqlalchemy/sql/operators.py
+++ b/lib/sqlalchemy/sql/operators.py
@@ -521,6 +521,11 @@ def nullslast_op(a):
 
 _commutative = set([eq, ne, add, mul])
 
+_comparison = set([eq, ne, lt, gt, ge, le])
+
+def is_comparison(op):
+    return op in _comparison
+
 def is_commutative(op):
     return op in _commutative
 
diff --git a/lib/sqlalchemy/sql/util.py b/lib/sqlalchemy/sql/util.py
index 8d2b5ecfd..cb8359048 100644
--- a/lib/sqlalchemy/sql/util.py
+++ b/lib/sqlalchemy/sql/util.py
@@ -62,6 +62,65 @@ def find_join_source(clauses, join_to):
     else:
         return None, None
 
+
+def visit_binary_product(fn, expr):
+    """Produce a traversal of the given expression, delivering
+    column comparisons to the given function.
+    
+    The function is of the form::
+    
+        def my_fn(binary, left, right)
+    
+    For each binary expression located which has a 
+    comparison operator, the product of "left" and
+    "right" will be delivered to that function,
+    in terms of that binary.
+    
+    Hence an expression like::
+    
+        and_(
+            (a + b) == q + func.sum(e + f),
+            j == r
+        )
+    
+    would have the traversal::
+    
+        a <eq> q
+        a <eq> e
+        a <eq> f
+        b <eq> q
+        b <eq> e
+        b <eq> f
+        j <eq> r
+
+    That is, every combination of "left" and
+    "right" that doesn't further contain
+    a binary comparison is passed as pairs.
+    
+    """
+    stack = []
+    def visit(element):
+        if isinstance(element, (expression._ScalarSelect)):
+            # we dont want to dig into correlated subqueries,
+            # those are just column elements by themselves
+            yield element
+        elif element.__visit_name__ == 'binary' and \
+            operators.is_comparison(element.operator):
+            stack.insert(0, element)
+            for l in visit(element.left):
+                for r in visit(element.right):
+                    fn(stack[0], l, r)
+            stack.pop(0)
+            for elem in element.get_children():
+                visit(elem)
+        else:
+            if isinstance(element, expression.ColumnClause):
+                yield element
+            for elem in element.get_children():
+                for e in visit(elem):
+                    yield e
+    list(visit(expr))
+
 def find_tables(clause, check_columns=False, 
                 include_aliases=False, include_joins=False, 
                 include_selects=False, include_crud=False):
@@ -225,7 +284,10 @@ def adapt_criterion_to_null(crit, nulls):
 
     return visitors.cloned_traverse(crit, {}, {'binary':visit_binary})
 
-def join_condition(a, b, ignore_nonexistent_tables=False, a_subset=None):
+
+def join_condition(a, b, ignore_nonexistent_tables=False, 
+                            a_subset=None,
+                            consider_as_foreign_keys=None):
     """create a join condition between two tables or selectables.
 
     e.g.::
@@ -261,6 +323,9 @@ def join_condition(a, b, ignore_nonexistent_tables=False, a_subset=None):
         for fk in sorted(
                     b.foreign_keys, 
                     key=lambda fk:fk.parent._creation_order):
+            if consider_as_foreign_keys is not None and \
+                fk.parent not in consider_as_foreign_keys:
+                continue
             try:
                 col = fk.get_referent(left)
             except exc.NoReferenceError, nrte:
@@ -276,6 +341,9 @@ def join_condition(a, b, ignore_nonexistent_tables=False, a_subset=None):
             for fk in sorted(
                         left.foreign_keys, 
                         key=lambda fk:fk.parent._creation_order):
+                if consider_as_foreign_keys is not None and \
+                    fk.parent not in consider_as_foreign_keys:
+                    continue
                 try:
                     col = fk.get_referent(b)
                 except exc.NoReferenceError, nrte:
@@ -298,11 +366,11 @@ def join_condition(a, b, ignore_nonexistent_tables=False, a_subset=None):
                                 "subquery using alias()?"
         else:
             hint = ""
-        raise exc.ArgumentError(
+        raise exc.NoForeignKeysError(
             "Can't find any foreign key relationships "
             "between '%s' and '%s'.%s" % (a.description, b.description, hint))
     elif len(constraints) > 1:
-        raise exc.ArgumentError(
+        raise exc.AmbiguousForeignKeysError(
             "Can't determine join between '%s' and '%s'; "
             "tables have more than one foreign key "
             "constraint relationship between them. "
@@ -356,13 +424,22 @@ class Annotated(object):
     def _annotate(self, values):
         _values = self._annotations.copy()
         _values.update(values)
+        return self._with_annotations(_values)
+
+    def _with_annotations(self, values):
         clone = self.__class__.__new__(self.__class__)
         clone.__dict__ = self.__dict__.copy()
-        clone._annotations = _values
+        clone._annotations = values
         return clone
 
-    def _deannotate(self):
-        return self.__element
+    def _deannotate(self, values=None, clone=True):
+        if values is None:
+            return self.__element
+        else:
+            _values = self._annotations.copy()
+            for v in values:
+                _values.pop(v, None)
+            return self._with_annotations(_values)
 
     def _compiler_dispatch(self, visitor, **kw):
         return self.__element.__class__._compiler_dispatch(self, visitor, **kw)
@@ -410,14 +487,8 @@ def _deep_annotate(element, annotations, exclude=None):
     Elements within the exclude collection will be cloned but not annotated.
 
     """
-    cloned = util.column_dict()
-
     def clone(elem):
-        # check if element is present in the exclude list.
-        # take into account proxying relationships.
-        if elem in cloned:
-            return cloned[elem]
-        elif exclude and \
+        if exclude and \
                     hasattr(elem, 'proxy_set') and \
                     elem.proxy_set.intersection(exclude):
             newelem = elem._clone()
@@ -426,24 +497,32 @@ def _deep_annotate(element, annotations, exclude=None):
         else:
             newelem = elem
         newelem._copy_internals(clone=clone)
-        cloned[elem] = newelem
         return newelem
 
     if element is not None:
         element = clone(element)
     return element
 
-def _deep_deannotate(element):
-    """Deep copy the given element, removing all annotations."""
+def _deep_deannotate(element, values=None):
+    """Deep copy the given element, removing annotations."""
 
     cloned = util.column_dict()
 
     def clone(elem):
-        if elem not in cloned:
-            newelem = elem._deannotate()
+        # if a values dict is given,
+        # the elem must be cloned each time it appears,
+        # as there may be different annotations in source
+        # elements that are remaining.  if totally
+        # removing all annotations, can assume the same
+        # slate...
+        if values or elem not in cloned:
+            newelem = elem._deannotate(values=values, clone=True)
             newelem._copy_internals(clone=clone)
-            cloned[elem] = newelem
-        return cloned[elem]
+            if not values:
+                cloned[elem] = newelem
+            return newelem
+        else:
+            return cloned[elem]
 
     if element is not None:
         element = clone(element)
@@ -547,6 +626,10 @@ def criterion_as_pairs(expression, consider_as_foreign_keys=None,
                                 "'consider_as_foreign_keys' or "
                                 "'consider_as_referenced_keys'")
 
+    def col_is(a, b):
+        #return a is b
+        return a.compare(b)
+
     def visit_binary(binary):
         if not any_operator and binary.operator is not operators.eq:
             return
@@ -556,20 +639,20 @@ def criterion_as_pairs(expression, consider_as_foreign_keys=None,
 
         if consider_as_foreign_keys:
             if binary.left in consider_as_foreign_keys and \
-                        (binary.right is binary.left or 
+                        (col_is(binary.right, binary.left) or 
                         binary.right not in consider_as_foreign_keys):
                 pairs.append((binary.right, binary.left))
             elif binary.right in consider_as_foreign_keys and \
-                        (binary.left is binary.right or 
+                        (col_is(binary.left, binary.right) or 
                         binary.left not in consider_as_foreign_keys):
                 pairs.append((binary.left, binary.right))
         elif consider_as_referenced_keys:
             if binary.left in consider_as_referenced_keys and \
-                        (binary.right is binary.left or 
+                        (col_is(binary.right, binary.left) or 
                         binary.right not in consider_as_referenced_keys):
                 pairs.append((binary.left, binary.right))
             elif binary.right in consider_as_referenced_keys and \
-                        (binary.left is binary.right or 
+                        (col_is(binary.left, binary.right) or 
                         binary.left not in consider_as_referenced_keys):
                 pairs.append((binary.right, binary.left))
         else:
@@ -681,11 +764,22 @@ class ClauseAdapter(visitors.ReplacingCloningVisitor):
       s.c.col1 == table2.c.col1
 
     """
-    def __init__(self, selectable, equivalents=None, include=None, exclude=None, adapt_on_names=False):
+    def __init__(self, selectable, equivalents=None, 
+                        include=None, exclude=None, 
+                        include_fn=None, exclude_fn=None, 
+                        adapt_on_names=False):
         self.__traverse_options__ = {'stop_on':[selectable]}
         self.selectable = selectable
-        self.include = include
-        self.exclude = exclude
+        if include:
+            assert not include_fn
+            self.include_fn = lambda e: e in include
+        else:
+            self.include_fn = include_fn
+        if exclude:
+            assert not exclude_fn
+            self.exclude_fn = lambda e: e in exclude
+        else:
+            self.exclude_fn = exclude_fn
         self.equivalents = util.column_dict(equivalents or {})
         self.adapt_on_names = adapt_on_names
 
@@ -705,19 +799,17 @@ class ClauseAdapter(visitors.ReplacingCloningVisitor):
         return newcol
 
     def replace(self, col):
-        if isinstance(col, expression.FromClause):
-            if self.selectable.is_derived_from(col):
+        if isinstance(col, expression.FromClause) and \
+            self.selectable.is_derived_from(col):
                 return self.selectable
-
-        if not isinstance(col, expression.ColumnElement):
+        elif not isinstance(col, expression.ColumnElement):
             return None
-
-        if self.include and col not in self.include:
+        elif self.include_fn and not self.include_fn(col):
             return None
-        elif self.exclude and col in self.exclude:
+        elif self.exclude_fn and self.exclude_fn(col):
             return None
-
-        return self._corresponding_column(col, True)
+        else:
+            return self._corresponding_column(col, True)
 
 class ColumnAdapter(ClauseAdapter):
     """Extends ClauseAdapter with extra utility functions.
diff --git a/lib/sqlalchemy/sql/visitors.py b/lib/sqlalchemy/sql/visitors.py
index 5354fbcbb..8a06982fc 100644
--- a/lib/sqlalchemy/sql/visitors.py
+++ b/lib/sqlalchemy/sql/visitors.py
@@ -242,13 +242,13 @@ def cloned_traverse(obj, opts, visitors):
         if elem in stop_on:
             return elem
         else:
-            if elem not in cloned:
-                cloned[elem] = newelem = elem._clone()
+            if id(elem) not in cloned:
+                cloned[id(elem)] = newelem = elem._clone()
                 newelem._copy_internals(clone=clone)
                 meth = visitors.get(newelem.__visit_name__, None)
                 if meth:
                     meth(newelem)
-            return cloned[elem]
+            return cloned[id(elem)]
 
     if obj is not None:
         obj = clone(obj)
@@ -260,16 +260,16 @@ def replacement_traverse(obj, opts, replace):
     replacement by a given replacement function."""
 
     cloned = util.column_dict()
-    stop_on = util.column_set(opts.get('stop_on', []))
+    stop_on = util.column_set([id(x) for x in opts.get('stop_on', [])])
 
     def clone(elem, **kw):
-        if elem in stop_on or \
+        if id(elem) in stop_on or \
             'no_replacement_traverse' in elem._annotations:
             return elem
         else:
             newelem = replace(elem)
             if newelem is not None:
-                stop_on.add(newelem)
+                stop_on.add(id(newelem))
                 return newelem
             else:
                 if elem not in cloned: