Fixed #12252 -- Ensure that queryset unions are commutative. Thanks to benreynwar for the report, and draft patch, and to Karen and Ramiro for the review eyeballs and patch updates.

git-svn-id: http://code.djangoproject.com/svn/django/trunk@15726 bcc190cf-cafb-0310-a4f2-bffc1f526a37

django/db/models/sql/query.py

@@ -446,6 +446,8 @@ class Query(object):
             "Cannot combine a unique query with a non-unique query."
 
         self.remove_inherited_models()
+        l_tables = set([a for a in self.tables if self.alias_refcount[a]])
+        r_tables = set([a for a in rhs.tables if rhs.alias_refcount[a]])
         # Work out how to relabel the rhs aliases, if necessary.
         change_map = {}
         used = set()
@@ -463,13 +465,19 @@ class Query(object):
             first = False
 
         # So that we don't exclude valid results in an "or" query combination,
-        # the first join that is exclusive to the lhs (self) must be converted
+        # all joins exclusive to either the lhs or the rhs must be converted
         # to an outer join.
         if not conjunction:
-            for alias in self.tables[1:]:
-                if self.alias_refcount[alias] == 1:
-                    self.promote_alias(alias, True)
-                    break
+            # Update r_tables aliases.
+            for alias in change_map:
+                if alias in r_tables:
+                    r_tables.remove(alias)
+                    r_tables.add(change_map[alias])
+            # Find aliases that are exclusive to rhs or lhs.
+            # These are promoted to outer joins.
+            outer_aliases = (l_tables | r_tables) - (l_tables & r_tables)
+            for alias in outer_aliases:
+                self.promote_alias(alias, True)
 
         # Now relabel a copy of the rhs where-clause and add it to the current
         # one.

tests/regressiontests/queries/models.py

@@ -294,3 +294,26 @@ class Node(models.Model):
 
     def __unicode__(self):
         return u"%s" % self.num
+
+# Bug #12252
+class ObjectA(models.Model):
+    name = models.CharField(max_length=50)
+
+    def __unicode__(self):
+        return self.name
+
+class ObjectB(models.Model):
+    name = models.CharField(max_length=50)
+    objecta = models.ForeignKey(ObjectA)
+    number = models.PositiveSmallIntegerField()
+
+    def __unicode__(self):
+        return self.name
+
+class ObjectC(models.Model):
+    name = models.CharField(max_length=50)
+    objecta = models.ForeignKey(ObjectA)
+    objectb = models.ForeignKey(ObjectB)
+
+    def __unicode__(self):
+       return self.name

tests/regressiontests/queries/tests.py

@@ -14,7 +14,8 @@ from django.utils.datastructures import SortedDict
 from models import (Annotation, Article, Author, Celebrity, Child, Cover, Detail,
     DumbCategory, ExtraInfo, Fan, Item, LeafA, LoopX, LoopZ, ManagedModel,
     Member, NamedCategory, Note, Number, Plaything, PointerA, Ranking, Related,
-    Report, ReservedName, Tag, TvChef, Valid, X, Food, Eaten, Node)
+    Report, ReservedName, Tag, TvChef, Valid, X, Food, Eaten, Node, ObjectA, ObjectB,
+    ObjectC)
 
 
 class BaseQuerysetTest(TestCase):
@@ -1658,3 +1659,63 @@ class ConditionalTests(BaseQuerysetTest):
             Number.objects.filter(num__in=numbers).count(),
             2500
         )
+
+class UnionTests(unittest.TestCase):
+    """
+    Tests for the union of two querysets. Bug #12252.
+    """
+    def setUp(self):
+        objectas = []
+        objectbs = []
+        objectcs = []
+        a_info = ['one', 'two', 'three']
+        for name in a_info:
+            o = ObjectA(name=name)
+            o.save()
+            objectas.append(o)
+        b_info = [('un', 1, objectas[0]), ('deux', 2, objectas[0]), ('trois', 3, objectas[2])]
+        for name, number, objecta in b_info:
+            o = ObjectB(name=name, number=number, objecta=objecta)
+            o.save()
+            objectbs.append(o)
+        c_info = [('ein', objectas[2], objectbs[2]), ('zwei', objectas[1], objectbs[1])]
+        for name, objecta, objectb in c_info:
+            o = ObjectC(name=name, objecta=objecta, objectb=objectb)
+            o.save()
+            objectcs.append(o)
+
+    def check_union(self, model, Q1, Q2):
+        filter = model.objects.filter
+        self.assertEqual(set(filter(Q1) | filter(Q2)), set(filter(Q1 | Q2)))
+        self.assertEqual(set(filter(Q2) | filter(Q1)), set(filter(Q1 | Q2)))
+
+    def test_A_AB(self):
+        Q1 = Q(name='two')
+        Q2 = Q(objectb__name='deux')
+        self.check_union(ObjectA, Q1, Q2)
+
+    def test_A_AB2(self):
+        Q1 = Q(name='two')
+        Q2 = Q(objectb__name='deux', objectb__number=2)
+        self.check_union(ObjectA, Q1, Q2)
+
+    def test_AB_ACB(self):
+        Q1 = Q(objectb__name='deux')
+        Q2 = Q(objectc__objectb__name='deux')
+        self.check_union(ObjectA, Q1, Q2)
+
+    def test_BAB_BAC(self):
+        Q1 = Q(objecta__objectb__name='deux')
+        Q2 = Q(objecta__objectc__name='ein')
+        self.check_union(ObjectB, Q1, Q2)
+
+    def test_BAB_BACB(self):
+        Q1 = Q(objecta__objectb__name='deux')
+        Q2 = Q(objecta__objectc__objectb__name='trois')
+        self.check_union(ObjectB, Q1, Q2)
+
+    def test_BA_BCA__BAB_BAC_BCA(self):
+        Q1 = Q(objecta__name='one', objectc__objecta__name='two')
+        Q2 = Q(objecta__objectc__name='ein', objectc__objecta__name='three', objecta__objectb__name='trois')
+        self.check_union(ObjectB, Q1, Q2)
+