Skip to content

HTTPS clone URL

Subversion checkout URL

You can clone with
or
.
Download ZIP
Browse files

Fixed #12252 -- Ensure that queryset unions are commutative. Thanks t…

…o 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
  • Loading branch information...
commit b7c41c1fbb2d45634dde5f7a450ba1a5aea5a8af 1 parent d1290b5
@freakboy3742 freakboy3742 authored
View
18 django/db/models/sql/query.py
@@ -446,6 +446,8 @@ def combine(self, rhs, connector):
"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 @@ def combine(self, rhs, connector):
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.
View
23 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
View
63 tests/regressiontests/queries/tests.py
@@ -14,7 +14,8 @@
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 @@ def test_ticket14244(self):
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)
+
Please sign in to comment.
Something went wrong with that request. Please try again.