Fixed #19385 again, now with real code changes

The commit of 266de5f9ae9e9f2fbfaec3b7e4b5fb9941967801 included only
tests, this time also code changes included...

5 files changed, 201 insertions, 115 deletions

diff --git a/django/db/models/sql/compiler.py b/django/db/models/sql/compiler.py
index 4711ea6e19..1f19131ba2 100644
--- a/django/db/models/sql/compiler.py
+++ b/django/db/models/sql/compiler.py
@@ -2,10 +2,9 @@ import datetime
 
 from django.conf import settings
 from django.core.exceptions import FieldError
-from django.db import transaction
 from django.db.backends.util import truncate_name
 from django.db.models.constants import LOOKUP_SEP
-from django.db.models.query_utils import select_related_descend
+from django.db.models.query_utils import select_related_descend, QueryWrapper
 from django.db.models.sql.constants import (SINGLE, MULTI, ORDER_DIR,
         GET_ITERATOR_CHUNK_SIZE, SelectInfo)
 from django.db.models.sql.datastructures import EmptyResultSet
@@ -33,7 +32,7 @@ class SQLCompiler(object):
         # cleaned. We are not using a clone() of the query here.
         """
         if not self.query.tables:
-            self.query.join((None, self.query.model._meta.db_table, None, None))
+            self.query.join((None, self.query.model._meta.db_table, None))
         if (not self.query.select and self.query.default_cols and not
                 self.query.included_inherited_models):
             self.query.setup_inherited_models()
@@ -273,7 +272,7 @@ class SQLCompiler(object):
         # be used by local fields.
         seen_models = {None: start_alias}
 
-        for field, model in opts.get_fields_with_model():
+        for field, model in opts.get_concrete_fields_with_model():
             if from_parent and model is not None and issubclass(from_parent, model):
                 # Avoid loading data for already loaded parents.
                 continue
@@ -314,9 +313,10 @@ class SQLCompiler(object):
 
         for name in self.query.distinct_fields:
             parts = name.split(LOOKUP_SEP)
-            field, col, alias, _, _ = self._setup_joins(parts, opts, None)
-            col, alias = self._final_join_removal(col, alias)
-            result.append("%s.%s" % (qn(alias), qn2(col)))
+            field, cols, alias, _, _ = self._setup_joins(parts, opts, None)
+            cols, alias = self._final_join_removal(cols, alias)
+            for col in cols:
+                result.append("%s.%s" % (qn(alias), qn2(col)))
         return result
 
 
@@ -387,15 +387,16 @@ class SQLCompiler(object):
             elif get_order_dir(field)[0] not in self.query.extra_select:
                 # 'col' is of the form 'field' or 'field1__field2' or
                 # '-field1__field2__field', etc.
-                for table, col, order in self.find_ordering_name(field,
+                for table, cols, order in self.find_ordering_name(field,
                         self.query.model._meta, default_order=asc):
-                    if (table, col) not in processed_pairs:
-                        elt = '%s.%s' % (qn(table), qn2(col))
-                        processed_pairs.add((table, col))
-                        if distinct and elt not in select_aliases:
-                            ordering_aliases.append(elt)
-                        result.append('%s %s' % (elt, order))
-                        group_by.append((elt, []))
+                    for col in cols:
+                        if (table, col) not in processed_pairs:
+                            elt = '%s.%s' % (qn(table), qn2(col))
+                            processed_pairs.add((table, col))
+                            if distinct and elt not in select_aliases:
+                                ordering_aliases.append(elt)
+                            result.append('%s %s' % (elt, order))
+                            group_by.append((elt, []))
             else:
                 elt = qn2(col)
                 if distinct and col not in select_aliases:
@@ -414,7 +415,7 @@ class SQLCompiler(object):
         """
         name, order = get_order_dir(name, default_order)
         pieces = name.split(LOOKUP_SEP)
-        field, col, alias, joins, opts = self._setup_joins(pieces, opts, alias)
+        field, cols, alias, joins, opts = self._setup_joins(pieces, opts, alias)
 
         # If we get to this point and the field is a relation to another model,
         # append the default ordering for that model.
@@ -432,8 +433,8 @@ class SQLCompiler(object):
                 results.extend(self.find_ordering_name(item, opts, alias,
                         order, already_seen))
             return results
-        col, alias = self._final_join_removal(col, alias)
-        return [(alias, col, order)]
+        cols, alias = self._final_join_removal(cols, alias)
+        return [(alias, cols, order)]
 
     def _setup_joins(self, pieces, opts, alias):
         """
@@ -446,13 +447,13 @@ class SQLCompiler(object):
         """
         if not alias:
             alias = self.query.get_initial_alias()
-        field, target, opts, joins, _ = self.query.setup_joins(
+        field, targets, opts, joins, _ = self.query.setup_joins(
             pieces, opts, alias)
         # We will later on need to promote those joins that were added to the
         # query afresh above.
         joins_to_promote = [j for j in joins if self.query.alias_refcount[j] < 2]
         alias = joins[-1]
-        col = target.column
+        cols = [target.column for target in targets]
         if not field.rel:
             # To avoid inadvertent trimming of a necessary alias, use the
             # refcount to show that we are referencing a non-relation field on
@@ -463,9 +464,9 @@ class SQLCompiler(object):
         # Ordering or distinct must not affect the returned set, and INNER
         # JOINS for nullable fields could do this.
         self.query.promote_joins(joins_to_promote)
-        return field, col, alias, joins, opts
+        return field, cols, alias, joins, opts
 
-    def _final_join_removal(self, col, alias):
+    def _final_join_removal(self, cols, alias):
         """
         A helper method for get_distinct and get_ordering. This method will
         trim extra not-needed joins from the tail of the join chain.
@@ -477,12 +478,14 @@ class SQLCompiler(object):
         if alias:
             while 1:
                 join = self.query.alias_map[alias]
-                if col != join.rhs_join_col:
+                lhs_cols, rhs_cols = zip(*[(lhs_col, rhs_col) for lhs_col, rhs_col in join.join_cols])
+                if set(cols) != set(rhs_cols):
                     break
+
+                cols = [lhs_cols[rhs_cols.index(col)] for col in cols]
                 self.query.unref_alias(alias)
                 alias = join.lhs_alias
-                col = join.lhs_join_col
-        return col, alias
+        return cols, alias
 
     def get_from_clause(self):
         """
@@ -504,22 +507,30 @@ class SQLCompiler(object):
             if not self.query.alias_refcount[alias]:
                 continue
             try:
-                name, alias, join_type, lhs, lhs_col, col, _, join_field = self.query.alias_map[alias]
+                name, alias, join_type, lhs, join_cols, _, join_field = self.query.alias_map[alias]
             except KeyError:
                 # Extra tables can end up in self.tables, but not in the
                 # alias_map if they aren't in a join. That's OK. We skip them.
                 continue
             alias_str = (alias != name and ' %s' % alias or '')
             if join_type and not first:
-                if join_field and hasattr(join_field, 'get_extra_join_sql'):
-                    extra_cond, extra_params = join_field.get_extra_join_sql(
-                        self.connection, qn, lhs, alias)
+                extra_cond = join_field.get_extra_restriction(
+                    self.query.where_class, alias, lhs)
+                if extra_cond:
+                    extra_sql, extra_params = extra_cond.as_sql(
+                        qn, self.connection)
+                    extra_sql = 'AND (%s)' % extra_sql
                     from_params.extend(extra_params)
                 else:
-                    extra_cond = ""
-                result.append('%s %s%s ON (%s.%s = %s.%s%s)' %
-                              (join_type, qn(name), alias_str, qn(lhs),
-                               qn2(lhs_col), qn(alias), qn2(col), extra_cond))
+                    extra_sql = ""
+                result.append('%s %s%s ON ('
+                        % (join_type, qn(name), alias_str))
+                for index, (lhs_col, rhs_col) in enumerate(join_cols):
+                    if index != 0:
+                        result.append(' AND ')
+                    result.append('%s.%s = %s.%s' %
+                    (qn(lhs), qn2(lhs_col), qn(alias), qn2(rhs_col)))
+                result.append('%s)' % extra_sql)
             else:
                 connector = not first and ', ' or ''
                 result.append('%s%s%s' % (connector, qn(name), alias_str))
@@ -545,7 +556,7 @@ class SQLCompiler(object):
             select_cols = self.query.select + self.query.related_select_cols
             # Just the column, not the fields.
             select_cols = [s[0] for s in select_cols]
-            if (len(self.query.model._meta.fields) == len(self.query.select)
+            if (len(self.query.model._meta.concrete_fields) == len(self.query.select)
                     and self.connection.features.allows_group_by_pk):
                 self.query.group_by = [
                     (self.query.model._meta.db_table, self.query.model._meta.pk.column)
@@ -623,14 +634,13 @@ class SQLCompiler(object):
             table = f.rel.to._meta.db_table
             promote = nullable or f.null
             alias = self.query.join_parent_model(opts, model, root_alias, {})
-
-            alias = self.query.join((alias, table, f.column,
-                    f.rel.get_related_field().column),
+            join_cols = f.get_joining_columns()
+            alias = self.query.join((alias, table, join_cols),
                     outer_if_first=promote, join_field=f)
             columns, aliases = self.get_default_columns(start_alias=alias,
                     opts=f.rel.to._meta, as_pairs=True)
             self.query.related_select_cols.extend(
-                SelectInfo(col, field) for col, field in zip(columns, f.rel.to._meta.fields))
+                SelectInfo(col, field) for col, field in zip(columns, f.rel.to._meta.concrete_fields))
             if restricted:
                 next = requested.get(f.name, {})
             else:
@@ -653,7 +663,7 @@ class SQLCompiler(object):
                 alias = self.query.join_parent_model(opts, f.rel.to, root_alias, {})
                 table = model._meta.db_table
                 alias = self.query.join(
-                    (alias, table, f.rel.get_related_field().column, f.column),
+                    (alias, table, f.get_joining_columns(reverse_join=True)),
                     outer_if_first=True, join_field=f
                 )
                 from_parent = (opts.model if issubclass(model, opts.model)
@@ -662,7 +672,7 @@ class SQLCompiler(object):
                     opts=model._meta, as_pairs=True, from_parent=from_parent)
                 self.query.related_select_cols.extend(
                     SelectInfo(col, field) for col, field
-                    in zip(columns, model._meta.fields))
+                    in zip(columns, model._meta.concrete_fields))
                 next = requested.get(f.related_query_name(), {})
                 # Use True here because we are looking at the _reverse_ side of
                 # the relation, which is always nullable.
@@ -706,7 +716,7 @@ class SQLCompiler(object):
                         if self.query.select:
                             fields = [f.field for f in self.query.select]
                         else:
-                            fields = self.query.model._meta.fields
+                            fields = self.query.model._meta.concrete_fields
                         fields = fields + [f.field for f in self.query.related_select_cols]
 
                         # If the field was deferred, exclude it from being passed
@@ -776,6 +786,22 @@ class SQLCompiler(object):
             return list(result)
         return result
 
+    def as_subquery_condition(self, alias, columns):
+        qn = self.quote_name_unless_alias
+        qn2 = self.connection.ops.quote_name
+        if len(columns) == 1:
+            sql, params = self.as_sql()
+            return '%s.%s IN (%s)' % (qn(alias), qn2(columns[0]), sql), params
+
+        for index, select_col in enumerate(self.query.select):
+            lhs = '%s.%s' % (qn(select_col.col[0]), qn2(select_col.col[1]))
+            rhs = '%s.%s' % (qn(alias), qn2(columns[index]))
+            self.query.where.add(
+                QueryWrapper('%s = %s' % (lhs, rhs), []), 'AND')
+
+        sql, params = self.as_sql()
+        return 'EXISTS (%s)' % sql, params
+
 
 class SQLInsertCompiler(SQLCompiler):
     def placeholder(self, field, val):
diff --git a/django/db/models/sql/constants.py b/django/db/models/sql/constants.py
index 81bd646d69..904f7b2c8b 100644
--- a/django/db/models/sql/constants.py
+++ b/django/db/models/sql/constants.py
@@ -25,7 +25,7 @@ GET_ITERATOR_CHUNK_SIZE = 100
 # dictionary in the Query class).
 JoinInfo = namedtuple('JoinInfo',
                       'table_name rhs_alias join_type lhs_alias '
-                      'lhs_join_col rhs_join_col nullable join_field')
+                      'join_cols nullable join_field')
 
 # Pairs of column clauses to select, and (possibly None) field for the clause.
 SelectInfo = namedtuple('SelectInfo', 'col field')
diff --git a/django/db/models/sql/expressions.py b/django/db/models/sql/expressions.py
index 389099161a..62adf79d87 100644
--- a/django/db/models/sql/expressions.py
+++ b/django/db/models/sql/expressions.py
@@ -55,13 +55,14 @@ class SQLEvaluator(object):
             self.cols.append((node, query.aggregate_select[node.name]))
         else:
             try:
-                field, source, opts, join_list, path = query.setup_joins(
+                field, sources, opts, join_list, path = query.setup_joins(
                     field_list, query.get_meta(),
                     query.get_initial_alias(), self.reuse)
-                target, _, join_list = query.trim_joins(source, join_list, path)
+                targets, _, join_list = query.trim_joins(sources, join_list, path)
                 if self.reuse is not None:
                     self.reuse.update(join_list)
-                self.cols.append((node, (join_list[-1], target.column)))
+                for t in targets:
+                    self.cols.append((node, (join_list[-1], t.column)))
             except FieldDoesNotExist:
                 raise FieldError("Cannot resolve keyword %r into field. "
                                  "Choices are: %s" % (self.name,
diff --git a/django/db/models/sql/query.py b/django/db/models/sql/query.py
index 2953d8cdaf..fb42cfc5db 100644
--- a/django/db/models/sql/query.py
+++ b/django/db/models/sql/query.py
@@ -452,13 +452,13 @@ class Query(object):
         # Now, add the joins from rhs query into the new query (skipping base
         # table).
         for alias in rhs.tables[1:]:
-            table, _, join_type, lhs, lhs_col, col, nullable, join_field = rhs.alias_map[alias]
+            table, _, join_type, lhs, join_cols, nullable, join_field = rhs.alias_map[alias]
             promote = (join_type == self.LOUTER)
             # If the left side of the join was already relabeled, use the
             # updated alias.
             lhs = change_map.get(lhs, lhs)
             new_alias = self.join(
-                (lhs, table, lhs_col, col), reuse=reuse,
+                (lhs, table, join_cols), reuse=reuse,
                 outer_if_first=not conjunction, nullable=nullable,
                 join_field=join_field)
             if promote:
@@ -682,7 +682,7 @@ class Query(object):
         aliases = list(aliases)
         while aliases:
             alias = aliases.pop(0)
-            if self.alias_map[alias].rhs_join_col is None:
+            if self.alias_map[alias].join_cols[0][1] is None:
                 # This is the base table (first FROM entry) - this table
                 # isn't really joined at all in the query, so we should not
                 # alter its join type.
@@ -818,7 +818,7 @@ class Query(object):
             alias = self.tables[0]
             self.ref_alias(alias)
         else:
-            alias = self.join((None, self.model._meta.db_table, None, None))
+            alias = self.join((None, self.model._meta.db_table, None))
         return alias
 
     def count_active_tables(self):
@@ -834,11 +834,12 @@ class Query(object):
         """
         Returns an alias for the join in 'connection', either reusing an
         existing alias for that join or creating a new one. 'connection' is a
-        tuple (lhs, table, lhs_col, col) where 'lhs' is either an existing
-        table alias or a table name. The join correspods to the SQL equivalent
-        of::
+        tuple (lhs, table, join_cols) where 'lhs' is either an existing
+        table alias or a table name. 'join_cols' is a tuple of tuples containing
+        columns to join on ((l_id1, r_id1), (l_id2, r_id2)). The join corresponds
+        to the SQL equivalent of::
 
-            lhs.lhs_col = table.col
+            lhs.l_id1 = table.r_id1 AND lhs.l_id2 = table.r_id2
 
         The 'reuse' parameter can be either None which means all joins
         (matching the connection) are reusable, or it can be a set containing
@@ -855,7 +856,7 @@ class Query(object):
 
         The 'join_field' is the field we are joining along (if any).
         """
-        lhs, table, lhs_col, col = connection
+        lhs, table, join_cols = connection
         assert lhs is None or join_field is not None
         existing = self.join_map.get(connection, ())
         if reuse is None:
@@ -884,7 +885,7 @@ class Query(object):
             join_type = self.LOUTER
         else:
             join_type = self.INNER
-        join = JoinInfo(table, alias, join_type, lhs, lhs_col, col, nullable,
+        join = JoinInfo(table, alias, join_type, lhs, join_cols or ((None, None),), nullable,
                         join_field)
         self.alias_map[alias] = join
         if connection in self.join_map:
@@ -941,7 +942,7 @@ class Query(object):
                 continue
             link_field = int_opts.get_ancestor_link(int_model)
             int_opts = int_model._meta
-            connection = (alias, int_opts.db_table, link_field.column, int_opts.pk.column)
+            connection = (alias, int_opts.db_table, link_field.get_joining_columns())
             alias = seen[int_model] = self.join(connection, nullable=False,
                                                 join_field=link_field)
         return alias or seen[None]
@@ -982,18 +983,20 @@ class Query(object):
             #   - this is an annotation over a model field
             # then we need to explore the joins that are required.
 
-            field, source, opts, join_list, path = self.setup_joins(
+            field, sources, opts, join_list, path = self.setup_joins(
                 field_list, opts, self.get_initial_alias())
 
             # Process the join chain to see if it can be trimmed
-            target, _, join_list = self.trim_joins(source, join_list, path)
+            targets, _, join_list = self.trim_joins(sources, join_list, path)
 
             # If the aggregate references a model or field that requires a join,
             # those joins must be LEFT OUTER - empty join rows must be returned
             # in order for zeros to be returned for those aggregates.
             self.promote_joins(join_list, True)
 
-            col = (join_list[-1], target.column)
+            col = targets[0].column
+            source = sources[0]
+            col = (join_list[-1], col)
         else:
             # The simplest cases. No joins required -
             # just reference the provided column alias.
@@ -1086,7 +1089,7 @@ class Query(object):
         allow_many = not branch_negated
 
         try:
-            field, target, opts, join_list, path = self.setup_joins(
+            field, sources, opts, join_list, path = self.setup_joins(
                     parts, opts, alias, can_reuse, allow_many,
                     allow_explicit_fk=True)
             if can_reuse is not None:
@@ -1106,13 +1109,19 @@ class Query(object):
         # the far end (fewer tables in a query is better). Note that join
         # promotion must happen before join trimming to have the join type
         # information available when reusing joins.
-        target, alias, join_list = self.trim_joins(target, join_list, path)
-        clause.add((Constraint(alias, target.column, field), lookup_type, value),
-                   AND)
+        targets, alias, join_list = self.trim_joins(sources, join_list, path)
+
+        if hasattr(field, 'get_lookup_constraint'):
+            constraint = field.get_lookup_constraint(self.where_class, alias, targets, sources,
+                                                     lookup_type, value)
+        else:
+            constraint = (Constraint(alias, targets[0].column, field), lookup_type, value)
+        clause.add(constraint, AND)
         if current_negated and (lookup_type != 'isnull' or value is False):
             self.promote_joins(join_list)
             if (lookup_type != 'isnull' and (
-                    self.is_nullable(target) or self.alias_map[join_list[-1]].join_type == self.LOUTER)):
+                    self.is_nullable(targets[0]) or
+                    self.alias_map[join_list[-1]].join_type == self.LOUTER)):
                 # The condition added here will be SQL like this:
                 # NOT (col IS NOT NULL), where the first NOT is added in
                 # upper layers of code. The reason for addition is that if col
@@ -1122,7 +1131,7 @@ class Query(object):
                 # (col IS NULL OR col != someval)
                 #   <=>
                 # NOT (col IS NOT NULL AND col = someval).
-                clause.add((Constraint(alias, target.column, None), 'isnull', False), AND)
+                clause.add((Constraint(alias, targets[0].column, None), 'isnull', False), AND)
         return clause
 
     def add_filter(self, filter_clause):
@@ -1272,22 +1281,26 @@ class Query(object):
                         opts = int_model._meta
                     else:
                         final_field = opts.parents[int_model]
-                        target = final_field.rel.get_related_field()
+                        targets = (final_field.rel.get_related_field(),)
                         opts = int_model._meta
-                        path.append(PathInfo(final_field, target, final_field.model._meta,
-                                             opts, final_field, False, True))
+                        path.append(PathInfo(final_field.model._meta, opts, targets, final_field, False, True))
             if hasattr(field, 'get_path_info'):
-                pathinfos, opts, target, final_field = field.get_path_info()
+                pathinfos = field.get_path_info()
                 if not allow_many:
                     for inner_pos, p in enumerate(pathinfos):
                         if p.m2m:
                             names_with_path.append((name, pathinfos[0:inner_pos + 1]))
                             raise MultiJoin(pos + 1, names_with_path)
+                last = pathinfos[-1]
                 path.extend(pathinfos)
+                final_field = last.join_field
+                opts = last.to_opts
+                targets = last.target_fields
                 names_with_path.append((name, pathinfos))
             else:
                 # Local non-relational field.
-                final_field = target = field
+                final_field = field
+                targets = (field,)
                 break
 
         if pos != len(names) - 1:
@@ -1297,7 +1310,7 @@ class Query(object):
                     "the lookup type?" % (name, names[pos + 1]))
             else:
                 raise FieldError("Join on field %r not permitted." % name)
-        return path, final_field, target
+        return path, final_field, targets
 
     def setup_joins(self, names, opts, alias, can_reuse=None, allow_many=True,
                     allow_explicit_fk=False):
@@ -1330,7 +1343,7 @@ class Query(object):
         """
         joins = [alias]
         # First, generate the path for the names
-        path, final_field, target = self.names_to_path(
+        path, final_field, targets = self.names_to_path(
             names, opts, allow_many, allow_explicit_fk)
         # Then, add the path to the query's joins. Note that we can't trim
         # joins at this stage - we will need the information about join type
@@ -1338,17 +1351,19 @@ class Query(object):
         for pos, join in enumerate(path):
             opts = join.to_opts
             if join.direct:
-                nullable = self.is_nullable(join.from_field)
+                nullable = self.is_nullable(join.join_field)
             else:
                 nullable = True
-            connection = alias, opts.db_table, join.from_field.column, join.to_field.column
+            connection = alias, opts.db_table, join.join_field.get_joining_columns()
             reuse = can_reuse if join.m2m else None
             alias = self.join(connection, reuse=reuse,
                               nullable=nullable, join_field=join.join_field)
             joins.append(alias)
-        return final_field, target, opts, joins, path
+        if hasattr(final_field, 'field'):
+            final_field = final_field.field
+        return final_field, targets, opts, joins, path
 
-    def trim_joins(self, target, joins, path):
+    def trim_joins(self, targets, joins, path):
         """
         The 'target' parameter is the final field being joined to, 'joins'
         is the full list of join aliases. The 'path' contain the PathInfos
@@ -1362,13 +1377,16 @@ class Query(object):
         trimmed as we don't know if there is anything on the other side of
         the join.
         """
-        for info in reversed(path):
-            if info.to_field == target and info.direct:
-                target = info.from_field
-                self.unref_alias(joins.pop())
-            else:
+        for pos, info in enumerate(reversed(path)):
+            if len(joins) == 1 or not info.direct:
                 break
-        return target, joins[-1], joins
+            join_targets = set(t.column for t in info.join_field.foreign_related_fields)
+            cur_targets = set(t.column for t in targets)
+            if not cur_targets.issubset(join_targets):
+                break
+            targets = tuple(r[0] for r in info.join_field.related_fields if r[1].column in cur_targets)
+            self.unref_alias(joins.pop())
+        return targets, joins[-1], joins
 
     def split_exclude(self, filter_expr, prefix, can_reuse, names_with_path):
         """
@@ -1413,17 +1431,31 @@ class Query(object):
         trimmed_prefix = []
         paths_in_prefix = trimmed_joins
         for name, path in names_with_path:
-            if paths_in_prefix - len(path) > 0:
-                trimmed_prefix.append(name)
-                paths_in_prefix -= len(path)
-            else:
-                trimmed_prefix.append(
-                    path[paths_in_prefix - len(path)].from_field.name)
+            if paths_in_prefix - len(path) < 0:
                 break
+            trimmed_prefix.append(name)
+            paths_in_prefix -= len(path)
+        join_field = path[paths_in_prefix].join_field
+        # TODO: This should be made properly multicolumn
+        # join aware. It is likely better to not use build_filter
+        # at all, instead construct joins up to the correct point,
+        # then construct the needed equality constraint manually,
+        # or maybe using SubqueryConstraint would work, too.
+        # The foreign_related_fields attribute is right here, we
+        # don't ever split joins for direct case.
+        trimmed_prefix.append(
+            join_field.field.foreign_related_fields[0].name)
         trimmed_prefix = LOOKUP_SEP.join(trimmed_prefix)
-        return self.build_filter(
+        condition = self.build_filter(
             ('%s__in' % trimmed_prefix, query),
             current_negated=True, branch_negated=True, can_reuse=can_reuse)
+        # Intentionally leave the other alias as blank, if the condition
+        # refers it, things will break here.
+        extra_restriction = join_field.get_extra_restriction(
+            self.where_class, None, [t for t in query.tables if query.alias_refcount[t]][0])
+        if extra_restriction:
+            query.where.add(extra_restriction, 'AND')
+        return condition
 
     def set_empty(self):
         self.where = EmptyWhere()
@@ -1502,20 +1534,17 @@ class Query(object):
 
         try:
             for name in field_names:
-                field, target, u2, joins, u3 = self.setup_joins(
+                field, targets, u2, joins, path = self.setup_joins(
                         name.split(LOOKUP_SEP), opts, alias, None, allow_m2m,
                         True)
-                final_alias = joins[-1]
-                col = target.column
-                if len(joins) > 1:
-                    join = self.alias_map[final_alias]
-                    if col == join.rhs_join_col:
-                        self.unref_alias(final_alias)
-                        final_alias = join.lhs_alias
-                        col = join.lhs_join_col
-                        joins = joins[:-1]
+
+                # Trim last join if possible
+                targets, final_alias, remaining_joins = self.trim_joins(targets, joins[-2:], path)
+                joins = joins[:-2] + remaining_joins
+
                 self.promote_joins(joins[1:])
-                self.select.append(SelectInfo((final_alias, col), field))
+                for target in targets:
+                    self.select.append(SelectInfo((final_alias, target.column), target))
         except MultiJoin:
             raise FieldError("Invalid field name: '%s'" % name)
         except FieldError:
@@ -1590,7 +1619,7 @@ class Query(object):
             opts = self.model._meta
             if not self.select:
                 count = self.aggregates_module.Count(
-                    (self.join((None, opts.db_table, None, None)), opts.pk.column),
+                    (self.join((None, opts.db_table, None)), opts.pk.column),
                     is_summary=True, distinct=True)
             else:
                 # Because of SQL portability issues, multi-column, distinct
@@ -1792,22 +1821,27 @@ class Query(object):
         in "WHERE somecol IN (subquery)". This construct is needed by
         split_exclude().
         _"""
-        join_pos = 0
+        all_paths = []
         for _, paths in names_with_path:
-            for path in paths:
-                peek = self.tables[join_pos + 1]
-                if self.alias_map[peek].join_type == self.LOUTER:
-                    # Back up one level and break
-                    select_alias = self.tables[join_pos]
-                    select_field = path.from_field
-                    break
-                select_alias = self.tables[join_pos + 1]
-                select_field = path.to_field
-                self.unref_alias(self.tables[join_pos])
-                join_pos += 1
-        self.select = [SelectInfo((select_alias, select_field.column), select_field)]
+            all_paths.extend(paths)
+        direct_join = True
+        for pos, path in enumerate(all_paths):
+            if self.alias_map[self.tables[pos + 1]].join_type == self.LOUTER:
+                direct_join = False
+                pos -= 1
+                break
+            self.unref_alias(self.tables[pos])
+        if path.direct:
+            direct_join = not direct_join
+        join_side = 0 if direct_join else 1
+        select_alias = self.tables[pos + 1]
+        join_field = path.join_field
+        if hasattr(join_field, 'field'):
+            join_field = join_field.field
+        select_fields = [r[join_side] for r in join_field.related_fields]
+        self.select = [SelectInfo((select_alias, f.column), f) for f in select_fields]
         self.remove_inherited_models()
-        return join_pos
+        return pos
 
     def is_nullable(self, field):
         """
diff --git a/django/db/models/sql/where.py b/django/db/models/sql/where.py
index 42682c342e..c738c914d1 100644
--- a/django/db/models/sql/where.py
+++ b/django/db/models/sql/where.py
@@ -382,3 +382,28 @@ class Constraint(object):
             new.__class__ = self.__class__
             new.alias, new.col, new.field = change_map[self.alias], self.col, self.field
             return new
+
+class SubqueryConstraint(object):
+    def __init__(self, alias, columns, targets, query_object):
+        self.alias = alias
+        self.columns = columns
+        self.targets = targets
+        self.query_object = query_object
+
+    def as_sql(self, qn, connection):
+        query = self.query_object
+
+        # QuerySet was sent
+        if hasattr(query, 'values'):
+            # as_sql should throw if we are using a
+            # connection on another database
+            query._as_sql(connection=connection)
+            query = query.values(*self.targets).query
+
+        query_compiler = query.get_compiler(connection=connection)
+        return query_compiler.as_subquery_condition(self.alias, self.columns)
+
+    def relabeled_clone(self, relabels):
+        return self.__class__(
+            relabels.get(self.alias, self.alias),
+            self.columns, self.query_object)