Index: django/db/models/sql/subqueries.py
===================================================================
--- django/db/models/sql/subqueries.py	(revision 14291)
+++ django/db/models/sql/subqueries.py	(working copy)
@@ -26,16 +26,17 @@
         self.where = where
         self.get_compiler(using).execute_sql(None)
 
-    def delete_batch(self, pk_list, using):
+    def delete_batch(self, pk_list, using, field=None):
         """
         Set up and execute delete queries for all the objects in pk_list.
 
         More than one physical query may be executed if there are a
         lot of values in pk_list.
         """
+        if not field:
+            field = self.model._meta.pk
         for offset in range(0, len(pk_list), GET_ITERATOR_CHUNK_SIZE):
             where = self.where_class()
-            field = self.model._meta.pk
             where.add((Constraint(None, field.column, field), 'in',
                     pk_list[offset : offset + GET_ITERATOR_CHUNK_SIZE]), AND)
             self.do_query(self.model._meta.db_table, where, using=using)
@@ -67,20 +68,14 @@
                 related_updates=self.related_updates.copy(), **kwargs)
 
 
-    def clear_related(self, related_field, pk_list, using):
-        """
-        Set up and execute an update query that clears related entries for the
-        keys in pk_list.
-
-        This is used by the QuerySet.delete_objects() method.
-        """
+    def update_batch(self, pk_list, values, using):
+        pk_field = self.model._meta.pk
+        self.add_update_values(values)
         for offset in range(0, len(pk_list), GET_ITERATOR_CHUNK_SIZE):
             self.where = self.where_class()
-            f = self.model._meta.pk
-            self.where.add((Constraint(None, f.column, f), 'in',
+            self.where.add((Constraint(None, pk_field.column, pk_field), 'in',
                     pk_list[offset : offset + GET_ITERATOR_CHUNK_SIZE]),
                     AND)
-            self.values = [(related_field, None, None)]
             self.get_compiler(using).execute_sql(None)
 
     def add_update_values(self, values):
Index: django/db/models/base.py
===================================================================
--- django/db/models/base.py	(revision 14291)
+++ django/db/models/base.py	(working copy)
@@ -7,8 +7,9 @@
 from django.db.models.fields import AutoField, FieldDoesNotExist
 from django.db.models.fields.related import (OneToOneRel, ManyToOneRel,
     OneToOneField, add_lazy_relation)
-from django.db.models.query import delete_objects, Q
-from django.db.models.query_utils import CollectedObjects, DeferredAttribute
+from django.db.models.query import Q
+from django.db.models.query_utils import DeferredAttribute
+from django.db.models.deletion import Collector
 from django.db.models.options import Options
 from django.db import connections, router, transaction, DatabaseError, DEFAULT_DB_ALIAS
 from django.db.models import signals
@@ -561,100 +562,14 @@
 
     save_base.alters_data = True
 
-    def _collect_sub_objects(self, seen_objs, parent=None, nullable=False):
-        """
-        Recursively populates seen_objs with all objects related to this
-        object.
-
-        When done, seen_objs.items() will be in the format:
-            [(model_class, {pk_val: obj, pk_val: obj, ...}),
-             (model_class, {pk_val: obj, pk_val: obj, ...}), ...]
-        """
-        pk_val = self._get_pk_val()
-        if seen_objs.add(self.__class__, pk_val, self,
-                         type(parent), parent, nullable):
-            return
-
-        for related in self._meta.get_all_related_objects():
-            rel_opts_name = related.get_accessor_name()
-            if not related.field.rel.multiple:
-                try:
-                    sub_obj = getattr(self, rel_opts_name)
-                except ObjectDoesNotExist:
-                    pass
-                else:
-                    sub_obj._collect_sub_objects(seen_objs, self, related.field.null)
-            else:
-                # To make sure we can access all elements, we can't use the
-                # normal manager on the related object. So we work directly
-                # with the descriptor object.
-                for cls in self.__class__.mro():
-                    if rel_opts_name in cls.__dict__:
-                        rel_descriptor = cls.__dict__[rel_opts_name]
-                        break
-                else:
-                    # in the case of a hidden fkey just skip it, it'll get
-                    # processed as an m2m
-                    if not related.field.rel.is_hidden():
-                        raise AssertionError("Should never get here.")
-                    else:
-                        continue
-                delete_qs = rel_descriptor.delete_manager(self).all()
-                for sub_obj in delete_qs:
-                    sub_obj._collect_sub_objects(seen_objs, self, related.field.null)
-
-        for related in self._meta.get_all_related_many_to_many_objects():
-            if related.field.rel.through:
-                db = router.db_for_write(related.field.rel.through.__class__, instance=self)
-                opts = related.field.rel.through._meta
-                reverse_field_name = related.field.m2m_reverse_field_name()
-                nullable = opts.get_field(reverse_field_name).null
-                filters = {reverse_field_name: self}
-                for sub_obj in related.field.rel.through._base_manager.using(db).filter(**filters):
-                    sub_obj._collect_sub_objects(seen_objs, self, nullable)
-
-        for f in self._meta.many_to_many:
-            if f.rel.through:
-                db = router.db_for_write(f.rel.through.__class__, instance=self)
-                opts = f.rel.through._meta
-                field_name = f.m2m_field_name()
-                nullable = opts.get_field(field_name).null
-                filters = {field_name: self}
-                for sub_obj in f.rel.through._base_manager.using(db).filter(**filters):
-                    sub_obj._collect_sub_objects(seen_objs, self, nullable)
-            else:
-                # m2m-ish but with no through table? GenericRelation: cascade delete
-                for sub_obj in f.value_from_object(self).all():
-                    # Generic relations not enforced by db constraints, thus we can set
-                    # nullable=True, order does not matter
-                    sub_obj._collect_sub_objects(seen_objs, self, True)
-
-        # Handle any ancestors (for the model-inheritance case). We do this by
-        # traversing to the most remote parent classes -- those with no parents
-        # themselves -- and then adding those instances to the collection. That
-        # will include all the child instances down to "self".
-        parent_stack = [p for p in self._meta.parents.values() if p is not None]
-        while parent_stack:
-            link = parent_stack.pop()
-            parent_obj = getattr(self, link.name)
-            if parent_obj._meta.parents:
-                parent_stack.extend(parent_obj._meta.parents.values())
-                continue
-            # At this point, parent_obj is base class (no ancestor models). So
-            # delete it and all its descendents.
-            parent_obj._collect_sub_objects(seen_objs)
-
     def delete(self, using=None):
         using = using or router.db_for_write(self.__class__, instance=self)
         assert self._get_pk_val() is not None, "%s object can't be deleted because its %s attribute is set to None." % (self._meta.object_name, self._meta.pk.attname)
 
-        # Find all the objects than need to be deleted.
-        seen_objs = CollectedObjects()
-        self._collect_sub_objects(seen_objs)
+        collector = Collector()
+        collector.collect([self], using=using)
+        collector.delete(using=using)
 
-        # Actually delete the objects.
-        delete_objects(seen_objs, using)
-
     delete.alters_data = True
 
     def _get_FIELD_display(self, field):
Index: django/db/models/options.py
===================================================================
--- django/db/models/options.py	(revision 14291)
+++ django/db/models/options.py	(working copy)
@@ -370,7 +370,7 @@
                     cache[obj] = parent
                 else:
                     cache[obj] = model
-        for klass in get_models():
+        for klass in get_models(include_auto_created=True):
             for f in klass._meta.local_fields:
                 if f.rel and not isinstance(f.rel.to, str) and self == f.rel.to._meta:
                     cache[RelatedObject(f.rel.to, klass, f)] = None
Index: django/db/models/fields/related.py
===================================================================
--- django/db/models/fields/related.py	(revision 14291)
+++ django/db/models/fields/related.py	(working copy)
@@ -7,6 +7,7 @@
 from django.db.models.related import RelatedObject
 from django.db.models.query import QuerySet
 from django.db.models.query_utils import QueryWrapper
+from django.db.models.deletion import CASCADE
 from django.utils.encoding import smart_unicode
 from django.utils.translation import ugettext_lazy as _, string_concat, ungettext, ugettext
 from django.utils.functional import curry
@@ -733,8 +734,7 @@
         manager.add(*value)
 
 class ManyToOneRel(object):
-    def __init__(self, to, field_name, related_name=None,
-            limit_choices_to=None, lookup_overrides=None, parent_link=False):
+    def __init__(self, to, field_name, related_name=None, limit_choices_to=None, parent_link=False, on_delete=None):
         try:
             to._meta
         except AttributeError: # to._meta doesn't exist, so it must be RECURSIVE_RELATIONSHIP_CONSTANT
@@ -744,9 +744,9 @@
         if limit_choices_to is None:
             limit_choices_to = {}
         self.limit_choices_to = limit_choices_to
-        self.lookup_overrides = lookup_overrides or {}
         self.multiple = True
         self.parent_link = parent_link
+        self.on_delete = on_delete
 
     def is_hidden(self):
         "Should the related object be hidden?"
@@ -764,11 +764,11 @@
         return data[0]
 
 class OneToOneRel(ManyToOneRel):
-    def __init__(self, to, field_name, related_name=None,
-            limit_choices_to=None, lookup_overrides=None, parent_link=False):
+    def __init__(self, to, field_name, related_name=None, limit_choices_to=None, parent_link=False, on_delete=None):
         super(OneToOneRel, self).__init__(to, field_name,
                 related_name=related_name, limit_choices_to=limit_choices_to,
-                lookup_overrides=lookup_overrides, parent_link=parent_link)
+                parent_link=parent_link, on_delete=on_delete
+        )
         self.multiple = False
 
 class ManyToManyRel(object):
@@ -820,8 +820,9 @@
         kwargs['rel'] = rel_class(to, to_field,
             related_name=kwargs.pop('related_name', None),
             limit_choices_to=kwargs.pop('limit_choices_to', None),
-            lookup_overrides=kwargs.pop('lookup_overrides', None),
-            parent_link=kwargs.pop('parent_link', False))
+            parent_link=kwargs.pop('parent_link', False),
+            on_delete=kwargs.pop('on_delete', CASCADE),
+        )
         Field.__init__(self, **kwargs)
 
     def validate(self, value, model_instance):
Index: django/db/models/__init__.py
===================================================================
--- django/db/models/__init__.py	(revision 14291)
+++ django/db/models/__init__.py	(working copy)
@@ -11,6 +11,7 @@
 from django.db.models.fields.subclassing import SubfieldBase
 from django.db.models.fields.files import FileField, ImageField
 from django.db.models.fields.related import ForeignKey, OneToOneField, ManyToManyField, ManyToOneRel, ManyToManyRel, OneToOneRel
+from django.db.models.deletion import CASCADE, PROTECT, SET, SET_NULL, SET_DEFAULT, DO_NOTHING
 from django.db.models import signals
 
 # Admin stages.
Index: django/db/models/deletion.py
===================================================================
--- django/db/models/deletion.py	(revision 0)
+++ django/db/models/deletion.py	(revision 0)
@@ -0,0 +1,209 @@
+from django.utils.datastructures import SortedDict
+from django.utils.functional import wraps
+from django.db import connections, transaction, IntegrityError
+from django.db.models import signals, sql
+from django.db.models.sql.constants import GET_ITERATOR_CHUNK_SIZE
+
+def CASCADE(collector, field, sub_objs):
+    collector.collect(sub_objs, source=field.rel.to, source_attr=field.name, nullable=field.null)
+    if field.null:
+        # FIXME: there should be a connection feature indicating whether nullable related fields should be nulled out before deletion
+        collector.add_field_update(field, None, sub_objs)
+
+def PROTECT(collector, field, sub_objs):
+    msg = "Cannot delete some instances of model '%s' because they are referenced through a protected foreign key: '%s.%s'" % (
+        field.rel.to.__name__, sub_objs[0].__class__.__name__, field.name
+    )
+    raise IntegrityError(msg)
+
+def SET(value):
+    def set_on_delete(collector, field, sub_objs):
+        collector.add_field_update(field, value, sub_objs)
+    return set_on_delete
+
+def SET_NULL(collector, field, sub_objs):
+    collector.add_field_update(field, None, sub_objs)
+
+def SET_DEFAULT(collector, field, sub_objs):
+    collector.add_field_update(field, field.get_default(), sub_objs)
+
+def DO_NOTHING(collector, field, sub_objs):
+    pass
+
+def force_managed(func):
+    @wraps(func)
+    def decorated(*args, **kwargs):
+        if not transaction.is_managed():
+            transaction.enter_transaction_management()
+            forced_managed = True
+        else:
+            forced_managed = False
+        try:                    
+            func(*args, **kwargs)
+            if forced_managed:
+                transaction.commit()
+            else:
+                transaction.commit_unless_managed()
+        finally:
+            if forced_managed:
+                transaction.leave_transaction_management()
+    return decorated
+
+class Collector(object):
+    def __init__(self):
+        self.data = {} # {model: [instances]}
+        self.batches = {} # {model: {field: set([instances])}}
+        self.field_updates = {} # {model: {(field, value): set([instances])}}        
+        self.dependencies = {} # {model: set([models])}
+
+    def add(self, objs, source=None, nullable=False):
+        """
+        Adds 'objs' to the collection of objects to be deleted.
+        If the call is the result of a cascade, 'source' should be the model that caused it 
+        and 'nullable' should be set to True, if the relation can be null.
+        
+        Returns a list of all objects that were not already collected.
+        """
+        if not objs:
+            return []
+        new_objs = []
+        model = objs[0].__class__
+        instances = self.data.setdefault(model, [])
+        for obj in objs:
+            if obj not in instances:
+                new_objs.append(obj)
+        instances.extend(new_objs)
+        # Nullable relationships can be ignored -- they are nulled out before
+        # deleting, and therefore do not affect the order in which objects
+        # have to be deleted.
+        if new_objs and source is not None and not nullable:
+            self.dependencies.setdefault(source, set()).add(model)
+        return new_objs
+        
+    def add_batch(self, model, field, objs):
+        """
+        Schedules a batch delete. Every instance of 'model' that is related to an instance of 'obj' through 'field' will be deleted.
+        """
+        self.batches.setdefault(model, {}).setdefault(field, set()).update(objs)
+        
+    def add_field_update(self, field, value, objs):
+        """
+        Schedules a field update. 'objs' must be a homogenous iterable collection of model instances (e.g. a QuerySet).
+        """
+        objs = list(objs)
+        if not objs:
+            return
+        model = objs[0].__class__
+        self.field_updates.setdefault(model, {}).setdefault((field, value), set()).update(objs)
+        
+    def collect(self, objs, source=None, nullable=False, collect_related=True, using=None, source_attr=None):
+        """
+        Adds 'objs' to the collection of objects to be deleted as well as all parent instances.
+        'objs' must be a homogenous iterable collection of model instances (e.g. a QuerySet).
+        If 'collect_related' is True, related objects will be handled by their respective on_delete handler.
+        
+        If the call is the result of a cascade, 'source' should be the model that caused it 
+        and 'nullable' should be set to True, if the relation can be null.
+        """
+        
+        new_objs = self.add(objs, source, nullable)
+        if not new_objs:
+            return
+        model = new_objs[0].__class__
+        
+        # Recusively collect parent models, but not their related objects.
+        for parent_model, ptr in model._meta.parents.iteritems():
+            if ptr:
+                parent_objs = [getattr(obj, ptr.name) for obj in new_objs]
+                self.collect(parent_objs, source=model, source_attr=ptr.rel.related_name, collect_related=False)
+
+        if collect_related:
+            for related in model._meta.get_all_related_objects():
+                field = related.field
+                if field.rel.is_hidden():
+                    self.add_batch(related.model, field, new_objs)
+                else:
+                    sub_objs = related.model._base_manager.using(using).filter(**{"%s__in" % field.name: new_objs})
+                    if not sub_objs:
+                        continue
+                    field.rel.on_delete(self, field, sub_objs)
+
+            # FIXME: support for generic relations should not require special handling
+            for field in model._meta.many_to_many:
+                if not field.rel.through:
+                    # m2m-ish but with no through table? GenericRelation: cascade delete
+                    for obj in new_objs:
+                        self.collect(field.value_from_object(obj).all(), source=model, source_attr=field.rel.related_name, nullable=True, using=using)
+
+    def instances_with_model(self):
+        for model, instances in self.data.iteritems():
+            for obj in instances:
+                yield model, obj
+                
+    def sort(self):
+        sorted_models = []
+        models = self.data.keys()
+        while len(sorted_models) < len(models):
+            found = False
+            for model in models:
+                if model in sorted_models:
+                    continue
+                dependencies = self.dependencies.get(model)
+                if not dependencies or not dependencies.difference(sorted_models):
+                    sorted_models.append(model)
+                    found = True
+            if not found:
+                return
+        self.data = SortedDict([(model, self.data[model]) for model in sorted_models])
+    
+    @force_managed
+    def delete(self, using=None):
+        # sort instance collections 
+        for instances in self.data.itervalues():
+            instances.sort(key=lambda obj: obj.pk)
+
+        # if possible, bring the models in an order suitable for databases that don't support transactions 
+        # or cannot defer contraint checks until the end of a transaction.
+        self.sort()
+        
+        # send pre_delete signals
+        for model, obj in self.instances_with_model():
+            if not model._meta.auto_created:
+                signals.pre_delete.send(sender=model, instance=obj, using=using)
+
+        # update fields
+        for model, instances_for_fieldvalues in self.field_updates.iteritems():
+            query = sql.UpdateQuery(model)
+            for (field, value), instances in instances_for_fieldvalues.iteritems():
+                query.update_batch([obj.pk for obj in instances], {field.name: value}, using)
+
+        # reverse instance collections
+        for instances in self.data.itervalues():
+            instances.reverse()
+
+        # delete batches
+        for model, batches in self.batches.iteritems():
+            query = sql.DeleteQuery(model)
+            for field, instances in batches.iteritems():
+                query.delete_batch([obj.pk for obj in instances], using, field)
+
+        # delete instances
+        for model, instances in self.data.iteritems():
+            query = sql.DeleteQuery(model)
+            pk_list = [obj.pk for obj in instances]
+            #query.delete_generic_relation_hack(pk_list, using)
+            query.delete_batch(pk_list, using)
+        
+        # send post_delete signals
+        for model, obj in self.instances_with_model():
+            if not model._meta.auto_created:
+                signals.post_delete.send(sender=model, instance=obj, using=using)
+        
+        # update collected instances
+        for model, instances_for_fieldvalues in self.field_updates.iteritems():
+            for (field, value), instances in instances_for_fieldvalues.iteritems():
+                for obj in instances:
+                    setattr(obj, field.attname, value)
+        for model, instances in self.data.iteritems():
+            for instance in instances:
+                setattr(instance, model._meta.pk.attname, None)
Index: django/db/models/query.py
===================================================================
--- django/db/models/query.py	(revision 14291)
+++ django/db/models/query.py	(working copy)
@@ -7,7 +7,8 @@
 from django.db import connections, router, transaction, IntegrityError
 from django.db.models.aggregates import Aggregate
 from django.db.models.fields import DateField
-from django.db.models.query_utils import Q, select_related_descend, CollectedObjects, CyclicDependency, deferred_class_factory, InvalidQuery
+from django.db.models.query_utils import Q, select_related_descend, deferred_class_factory, InvalidQuery
+from django.db.models.deletion import Collector
 from django.db.models import signals, sql
 from django.utils.copycompat import deepcopy
 
@@ -426,23 +427,10 @@
         del_query.query.select_related = False
         del_query.query.clear_ordering()
 
-        # Delete objects in chunks to prevent the list of related objects from
-        # becoming too long.
-        seen_objs = None
-        del_itr = iter(del_query)
-        while 1:
-            # Collect a chunk of objects to be deleted, and then all the
-            # objects that are related to the objects that are to be deleted.
-            # The chunking *isn't* done by slicing the del_query because we
-            # need to maintain the query cache on del_query (see #12328)
-            seen_objs = CollectedObjects(seen_objs)
-            for i, obj in izip(xrange(CHUNK_SIZE), del_itr):
-                obj._collect_sub_objects(seen_objs)
+        collector = Collector()
+        collector.collect(del_query)
+        collector.delete(using=del_query.db)
 
-            if not seen_objs:
-                break
-            delete_objects(seen_objs, del_query.db)
-
         # Clear the result cache, in case this QuerySet gets reused.
         self._result_cache = None
     delete.alters_data = True
Index: django/db/models/query_utils.py
===================================================================
--- django/db/models/query_utils.py	(revision 14291)
+++ django/db/models/query_utils.py	(working copy)
@@ -14,13 +14,6 @@
 from django.utils.datastructures import SortedDict
 
 
-class CyclicDependency(Exception):
-    """
-    An error when dealing with a collection of objects that have a cyclic
-    dependency, i.e. when deleting multiple objects.
-    """
-    pass
-
 class InvalidQuery(Exception):
     """
     The query passed to raw isn't a safe query to use with raw.
@@ -28,107 +21,6 @@
     pass
 
 
-class CollectedObjects(object):
-    """
-    A container that stores keys and lists of values along with remembering the
-    parent objects for all the keys.
-
-    This is used for the database object deletion routines so that we can
-    calculate the 'leaf' objects which should be deleted first.
-
-    previously_seen is an optional argument. It must be a CollectedObjects
-    instance itself; any previously_seen collected object will be blocked from
-    being added to this instance.
-    """
-
-    def __init__(self, previously_seen=None):
-        self.data = {}
-        self.children = {}
-        if previously_seen:
-            self.blocked = previously_seen.blocked
-            for cls, seen in previously_seen.data.items():
-                self.blocked.setdefault(cls, SortedDict()).update(seen)
-        else:
-            self.blocked = {}
-
-    def add(self, model, pk, obj, parent_model, parent_obj=None, nullable=False):
-        """
-        Adds an item to the container.
-
-        Arguments:
-        * model - the class of the object being added.
-        * pk - the primary key.
-        * obj - the object itself.
-        * parent_model - the model of the parent object that this object was
-          reached through.
-        * parent_obj - the parent object this object was reached
-          through (not used here, but needed in the API for use elsewhere)
-        * nullable - should be True if this relation is nullable.
-
-        Returns True if the item already existed in the structure and
-        False otherwise.
-        """
-        if pk in self.blocked.get(model, {}):
-            return True
-
-        d = self.data.setdefault(model, SortedDict())
-        retval = pk in d
-        d[pk] = obj
-        # Nullable relationships can be ignored -- they are nulled out before
-        # deleting, and therefore do not affect the order in which objects
-        # have to be deleted.
-        if parent_model is not None and not nullable:
-            self.children.setdefault(parent_model, []).append(model)
-        return retval
-
-    def __contains__(self, key):
-        return self.data.__contains__(key)
-
-    def __getitem__(self, key):
-        return self.data[key]
-
-    def __nonzero__(self):
-        return bool(self.data)
-
-    def iteritems(self):
-        for k in self.ordered_keys():
-            yield k, self[k]
-
-    def items(self):
-        return list(self.iteritems())
-
-    def keys(self):
-        return self.ordered_keys()
-
-    def ordered_keys(self):
-        """
-        Returns the models in the order that they should be dealt with (i.e.
-        models with no dependencies first).
-        """
-        dealt_with = SortedDict()
-        # Start with items that have no children
-        models = self.data.keys()
-        while len(dealt_with) < len(models):
-            found = False
-            for model in models:
-                if model in dealt_with:
-                    continue
-                children = self.children.setdefault(model, [])
-                if len([c for c in children if c not in dealt_with]) == 0:
-                    dealt_with[model] = None
-                    found = True
-            if not found:
-                raise CyclicDependency(
-                    "There is a cyclic dependency of items to be processed.")
-
-        return dealt_with.keys()
-
-    def unordered_keys(self):
-        """
-        Fallback for the case where is a cyclic dependency but we don't  care.
-        """
-        return self.data.keys()
-
 class QueryWrapper(object):
     """
     A type that indicates the contents are an SQL fragment and the associate
Index: django/core/management/validation.py
===================================================================
--- django/core/management/validation.py	(revision 14291)
+++ django/core/management/validation.py	(working copy)
@@ -22,6 +22,7 @@
     from django.db import models, connection
     from django.db.models.loading import get_app_errors
     from django.db.models.fields.related import RelatedObject
+    from django.db.models.deletion import SET_NULL, SET_DEFAULT
 
     e = ModelErrorCollection(outfile)
 
@@ -85,6 +86,13 @@
             # Perform any backend-specific field validation.
             connection.validation.validate_field(e, opts, f)
 
+            # Check if the on_delete behavior is sane
+            if f.rel and hasattr(f.rel, 'on_delete'):
+                if f.rel.on_delete == SET_NULL and not f.null:
+                    e.add(opts, "'%s' specifies on_delete=SET_NULL, but cannot be null." % f.name)
+                elif f.rel.on_delete == SET_DEFAULT and not f.has_default():
+                    e.add(opts, "'%s' specifies on_delete=SET_DEFAULT, but has no default value." % f.name)
+
             # Check to see if the related field will clash with any existing
             # fields, m2m fields, m2m related objects or related objects
             if f.rel:
Index: django/contrib/admin/util.py
===================================================================
--- django/contrib/admin/util.py	(revision 14291)
+++ django/contrib/admin/util.py	(working copy)
@@ -1,4 +1,5 @@
 from django.db import models
+from django.db.models.deletion import Collector
 from django.db.models.related import RelatedObject
 from django.forms.forms import pretty_name
 from django.utils import formats
@@ -104,13 +105,10 @@
     method uses this function also from a change_list view.
     This will not be used if we can reverse the URL.
     """
+    # FIXME: This code is broken. The old version depends on Model._collect_sub_objects() and contained a TODO comment.
     collector = NestedObjects()
-    for obj in objs:
-        # TODO using a private model API!
-        obj._collect_sub_objects(collector)
-
+    collector.collect(objs)
     perms_needed = set()
-
     to_delete = collector.nested(_format_callback,
                                  user=user,
                                  admin_site=admin_site,
@@ -120,83 +118,35 @@
     return to_delete, perms_needed
 
 
-class NestedObjects(object):
-    """
-    A directed acyclic graph collection that exposes the add() API
-    expected by Model._collect_sub_objects and can present its data as
-    a nested list of objects.
-
-    """
+class NestedObjects(Collector):
     def __init__(self):
-        # Use object keys of the form (model, pk) because actual model
-        # objects may not be unique
+        super(NestedObjects, self).__init__()
+        self.edges = {} # {from_instance: [to_instances]}
+        
+    def add_edge(self, source, target):
+        self.edges.setdefault(source, []).append(target)
+        
+    def collect(self, objs, source_attr=None, **kwargs):
+        for obj in objs:
+            if source_attr:
+                self.add_edge(getattr(obj, source_attr), obj)
+            else:
+                self.add_edge(None, obj)
+        return super(NestedObjects, self).collect(objs, source_attr=source_attr, **kwargs)
 
-        # maps object key to list of child keys
-        self.children = SortedDict()
-
-        # maps object key to parent key
-        self.parents = SortedDict()
-
-        # maps object key to actual object
-        self.seen = SortedDict()
-
-    def add(self, model, pk, obj,
-            parent_model=None, parent_obj=None, nullable=False):
-        """
-        Add item ``obj`` to the graph. Returns True (and does nothing)
-        if the item has been seen already.
-
-        The ``parent_obj`` argument must already exist in the graph; if
-        not, it's ignored (but ``obj`` is still added with no
-        parent). In any case, Model._collect_sub_objects (for whom
-        this API exists) will never pass a parent that hasn't already
-        been added itself.
-
-        These restrictions in combination ensure the graph will remain
-        acyclic (but can have multiple roots).
-
-        ``model``, ``pk``, and ``parent_model`` arguments are ignored
-        in favor of the appropriate lookups on ``obj`` and
-        ``parent_obj``; unlike CollectedObjects, we can't maintain
-        independence from the knowledge that we're operating on model
-        instances, and we don't want to allow for inconsistency.
-
-        ``nullable`` arg is ignored: it doesn't affect how the tree of
-        collected objects should be nested for display.
-        """
-        model, pk = type(obj), obj._get_pk_val()
-
-        # auto-created M2M models don't interest us
-        if model._meta.auto_created:
-            return True
-
-        key = model, pk
-
-        if key in self.seen:
-            return True
-        self.seen.setdefault(key, obj)
-
-        if parent_obj is not None:
-            parent_model, parent_pk = (type(parent_obj),
-                                       parent_obj._get_pk_val())
-            parent_key = (parent_model, parent_pk)
-            if parent_key in self.seen:
-                self.children.setdefault(parent_key, list()).append(key)
-                self.parents.setdefault(key, parent_key)
-
-    def _nested(self, key, format_callback=None, **kwargs):
-        obj = self.seen[key]
+    def _nested(self, obj, seen, format_callback, kwargs):
+        if obj in seen:
+            return []
+        seen.add(obj)
+        children = []
+        for child in self.edges.get(obj, ()):
+            children.extend(self._nested(child, seen, format_callback, kwargs))
         if format_callback:
             ret = [format_callback(obj, **kwargs)]
         else:
             ret = [obj]
-
-        children = []
-        for child in self.children.get(key, ()):
-            children.extend(self._nested(child, format_callback, **kwargs))
         if children:
             ret.append(children)
-
         return ret
 
     def nested(self, format_callback=None, **kwargs):
@@ -206,10 +156,10 @@
         Passes **kwargs back to the format_callback as kwargs.
 
         """
+        seen = set()
         roots = []
-        for key in self.seen.keys():
-            if key not in self.parents:
-                roots.extend(self._nested(key, format_callback, **kwargs))
+        for root in self.edges.get(None, ()):
+            roots.extend(self._nested(root, seen, format_callback, kwargs))
         return roots
 
 
Index: tests/modeltests/invalid_models/models.py
===================================================================
--- tests/modeltests/invalid_models/models.py	(revision 14291)
+++ tests/modeltests/invalid_models/models.py	(working copy)
@@ -207,6 +207,13 @@
     tgt = models.ForeignKey(FKTarget, to_field='good')
 
 
+class InvalidSetNull(models.Model):
+    fk = models.ForeignKey('self', on_delete=models.SET_NULL)
+
+class InvalidSetDefault(models.Model):
+    fk = models.ForeignKey('self', on_delete=models.SET_DEFAULT)
+
+
 model_errors = """invalid_models.fielderrors: "charfield": CharFields require a "max_length" attribute that is a positive integer.
 invalid_models.fielderrors: "charfield2": CharFields require a "max_length" attribute that is a positive integer.
 invalid_models.fielderrors: "charfield3": CharFields require a "max_length" attribute that is a positive integer.
@@ -311,4 +318,6 @@
 invalid_models.uniquem2m: ManyToManyFields cannot be unique.  Remove the unique argument on 'unique_people'.
 invalid_models.nonuniquefktarget1: Field 'bad' under model 'FKTarget' must have a unique=True constraint.
 invalid_models.nonuniquefktarget2: Field 'bad' under model 'FKTarget' must have a unique=True constraint.
+invalid_models.invalidsetnull: 'fk' specifies on_delete=SET_NULL, but cannot be null.
+invalid_models.invalidsetdefault: 'fk' specifies on_delete=SET_DEFAULT, but has no default value.
 """
Index: tests/modeltests/delete/tests.py
===================================================================
--- tests/modeltests/delete/tests.py	(revision 14291)
+++ tests/modeltests/delete/tests.py	(working copy)
@@ -1,7 +1,5 @@
 from django.db.models import sql
 from django.db.models.loading import cache
-from django.db.models.query import CollectedObjects
-from django.db.models.query_utils import CyclicDependency
 from django.test import TestCase
 
 from models import A, B, C, D, E, F
@@ -24,112 +22,3 @@
         self.order_models("a", "b", "c", "d", "e", "f")
         self.clear_rel_obj_caches(A, B, C, D, E, F)
 
-    def test_collected_objects(self):
-        g = CollectedObjects()
-        self.assertFalse(g.add("key1", 1, "item1", None))
-        self.assertEqual(g["key1"], {1: "item1"})
-
-        self.assertFalse(g.add("key2", 1, "item1", "key1"))
-        self.assertFalse(g.add("key2", 2, "item2", "key1"))
-
-        self.assertEqual(g["key2"], {1: "item1", 2: "item2"})
-
-        self.assertFalse(g.add("key3", 1, "item1", "key1"))
-        self.assertTrue(g.add("key3", 1, "item1", "key2"))
-        self.assertEqual(g.ordered_keys(), ["key3", "key2", "key1"])
-
-        self.assertTrue(g.add("key2", 1, "item1", "key3"))
-        self.assertRaises(CyclicDependency, g.ordered_keys)
-
-    def test_delete(self):
-        ## Second, test the usage of CollectedObjects by Model.delete()
-
-        # Due to the way that transactions work in the test harness, doing
-        # m.delete() here can work but fail in a real situation, since it may
-        # delete all objects, but not in the right order. So we manually check
-        # that the order of deletion is correct.
-
-        # Also, it is possible that the order is correct 'accidentally', due
-        # solely to order of imports etc.  To check this, we set the order that
-        # 'get_models()' will retrieve to a known 'nice' order, and then try
-        # again with a known 'tricky' order.  Slightly naughty access to
-        # internals here :-)
-
-        # If implementation changes, then the tests may need to be simplified:
-        #  - remove the lines that set the .keyOrder and clear the related
-        #    object caches
-        #  - remove the second set of tests (with a2, b2 etc)
-
-        a1 = A.objects.create()
-        b1 = B.objects.create(a=a1)
-        c1 = C.objects.create(b=b1)
-        d1 = D.objects.create(c=c1, a=a1)
-
-        o = CollectedObjects()
-        a1._collect_sub_objects(o)
-        self.assertEqual(o.keys(), [D, C, B, A])
-        a1.delete()
-
-        # Same again with a known bad order
-        self.order_models("d", "c", "b", "a")
-        self.clear_rel_obj_caches(A, B, C, D)
-
-        a2 = A.objects.create()
-        b2 = B.objects.create(a=a2)
-        c2 = C.objects.create(b=b2)
-        d2 = D.objects.create(c=c2, a=a2)
-
-        o = CollectedObjects()
-        a2._collect_sub_objects(o)
-        self.assertEqual(o.keys(), [D, C, B, A])
-        a2.delete()
-
-    def test_collected_objects_null(self):
-        g = CollectedObjects()
-        self.assertFalse(g.add("key1", 1, "item1", None))
-        self.assertFalse(g.add("key2", 1, "item1", "key1", nullable=True))
-        self.assertTrue(g.add("key1", 1, "item1", "key2"))
-        self.assertEqual(g.ordered_keys(), ["key1", "key2"])
-
-    def test_delete_nullable(self):
-        e1 = E.objects.create()
-        f1 = F.objects.create(e=e1)
-        e1.f = f1
-        e1.save()
-
-        # Since E.f is nullable, we should delete F first (after nulling out
-        # the E.f field), then E.
-
-        o = CollectedObjects()
-        e1._collect_sub_objects(o)
-        self.assertEqual(o.keys(), [F, E])
-
-        # temporarily replace the UpdateQuery class to verify that E.f is
-        # actually nulled out first
-
-        logged = []
-        class LoggingUpdateQuery(sql.UpdateQuery):
-            def clear_related(self, related_field, pk_list, using):
-                logged.append(related_field.name)
-                return super(LoggingUpdateQuery, self).clear_related(related_field, pk_list, using)
-        original = sql.UpdateQuery
-        sql.UpdateQuery = LoggingUpdateQuery
-
-        e1.delete()
-        self.assertEqual(logged, ["f"])
-        logged = []
-
-        e2 = E.objects.create()
-        f2 = F.objects.create(e=e2)
-        e2.f = f2
-        e2.save()
-
-        # Same deal as before, though we are starting from the other object.
-        o = CollectedObjects()
-        f2._collect_sub_objects(o)
-        self.assertEqual(o.keys(), [F, E])
-        f2.delete()
-        self.assertEqual(logged, ["f"])
-        logged = []
-
-        sql.UpdateQuery = original
Index: tests/modeltests/on_delete/__init__.py
===================================================================
--- tests/modeltests/on_delete/__init__.py	(revision 0)
+++ tests/modeltests/on_delete/__init__.py	(revision 0)
@@ -0,0 +1 @@
+#
\ No newline at end of file
Index: tests/modeltests/on_delete/models.py
===================================================================
--- tests/modeltests/on_delete/models.py	(revision 0)
+++ tests/modeltests/on_delete/models.py	(revision 0)
@@ -0,0 +1,216 @@
+from django.test import TestCase
+from django.db import models, IntegrityError
+
+class R(models.Model):
+    is_default = models.BooleanField(default=False)
+
+    def __str__(self):
+        return "%s" % self.pk
+
+get_default_r = lambda: R.objects.get_or_create(is_default=True)[0]
+    
+class S(models.Model):
+    r = models.ForeignKey(R)
+    
+class T(models.Model):
+    s = models.ForeignKey(S)
+
+class U(models.Model):
+    t = models.ForeignKey(T)
+
+
+class A(models.Model):
+    name = models.CharField(max_length=10)    
+
+    auto = models.ForeignKey(R, related_name="auto_set")
+    auto_nullable = models.ForeignKey(R, null=True, related_name='auto_nullable_set')
+    setnull = models.ForeignKey(R, on_delete=models.SET_NULL, null=True, related_name='setnull_set')
+    setdefault = models.ForeignKey(R, on_delete=models.SET_DEFAULT, default=get_default_r, related_name='setdefault_set')
+    setdefault_none = models.ForeignKey(R, on_delete=models.SET_DEFAULT, default=None, null=True, related_name='setnull_nullable_set')
+    cascade = models.ForeignKey(R, on_delete=models.CASCADE, related_name='cascade_set')
+    cascade_nullable = models.ForeignKey(R, on_delete=models.CASCADE, null=True, related_name='cascade_nullable_set')
+    protect = models.ForeignKey(R, on_delete=models.PROTECT, null=True)
+    donothing = models.ForeignKey(R, on_delete=models.DO_NOTHING, null=True, related_name='donothing_set')
+    
+def create_a(name):
+    a = A(name=name)
+    for name in ('auto', 'auto_nullable', 'setnull', 'setdefault', 'setdefault_none', 'cascade', 'cascade_nullable', 'protect', 'donothing'):
+        r = R.objects.create()
+        setattr(a, name, r)
+    a.save()
+    return a
+    
+class M(models.Model):
+    m2m = models.ManyToManyField(R, related_name="m_set")    
+    m2m_through = models.ManyToManyField(R, through="MR", related_name="m_through_set")
+    m2m_through_null = models.ManyToManyField(R, through="MRNull", related_name="m_through_null_set")
+    
+class MR(models.Model):
+    m = models.ForeignKey(M)
+    r = models.ForeignKey(R)
+
+class MRNull(models.Model):
+    m = models.ForeignKey(M)
+    r = models.ForeignKey(R, null=True, on_delete=models.SET_NULL)
+
+class OnDeleteTests(TestCase):
+    def test_basics(self):
+        DEFAULT = get_default_r()
+        
+        a = create_a('auto')
+        a.auto.delete()
+        self.failIf(A.objects.filter(name='auto').exists())
+        
+        a = create_a('auto_nullable')
+        a.auto_nullable.delete()
+        self.failIf(A.objects.filter(name='auto_nullable').exists())
+        
+        a = create_a('setnull')
+        a.setnull.delete()
+        a = A.objects.get(pk=a.pk)
+        self.failUnlessEqual(None, a.setnull)
+        
+        a = create_a('setdefault')
+        a.setdefault.delete()
+        a = A.objects.get(pk=a.pk)
+        self.failUnlessEqual(DEFAULT, a.setdefault)
+        
+        a = create_a('setdefault_none')
+        a.setdefault_none.delete()
+        a = A.objects.get(pk=a.pk)
+        self.failUnlessEqual(None, a.setdefault_none)
+        
+        a = create_a('cascade')
+        a.cascade.delete()
+        self.failIf(A.objects.filter(name='cascade').exists())
+        
+        a = create_a('cascade_nullable')
+        a.cascade_nullable.delete()
+        self.failIf(A.objects.filter(name='cascade_nullable').exists())
+        
+        a = create_a('protect')
+        self.assertRaises(IntegrityError, a.protect.delete)
+        
+        # Testing DO_NOTHING is a bit harder: It would raise IntegrityError for a normal model, 
+        # so we connect to pre_delete and set the fk to a known value.
+        replacement_r = R.objects.create()
+        def check_do_nothing(sender, **kwargs):
+            obj = kwargs['instance']
+            obj.donothing_set.update(donothing=replacement_r)
+        models.signals.pre_delete.connect(check_do_nothing)
+        a = create_a('do_nothing')
+        a.donothing.delete()
+        a = A.objects.get(pk=a.pk)
+        self.failUnlessEqual(replacement_r, a.donothing)
+        models.signals.pre_delete.disconnect(check_do_nothing)        
+        
+        A.objects.all().update(protect=None, donothing=None)
+        R.objects.all().delete()
+        self.failIf(A.objects.exists())
+        
+    def test_m2m(self):
+        m = M.objects.create()
+        r = R.objects.create()
+        MR.objects.create(m=m, r=r)
+        r.delete()
+        self.failIf(MR.objects.exists())
+        
+        r = R.objects.create()
+        MR.objects.create(m=m, r=r)
+        m.delete()
+        self.failIf(MR.objects.exists())
+        
+        m = M.objects.create()
+        r = R.objects.create()
+        m.m2m.add(r)
+        r.delete()
+        through = M._meta.get_field('m2m').rel.through
+        self.failIf(through.objects.exists())
+        
+        r = R.objects.create()
+        m.m2m.add(r)
+        m.delete()
+        self.failIf(through.objects.exists())
+        
+        m = M.objects.create()
+        r = R.objects.create()
+        MRNull.objects.create(m=m, r=r)
+        r.delete()
+        self.failIf(not MRNull.objects.exists())
+        self.failIf(m.m2m_through_null.exists())
+        
+    
+    def assert_num_queries(self, num, func, *args, **kwargs):
+        # FIXME: replace with the new builtin method
+        from django.conf import settings
+        from django.db import connection
+        old_debug = settings.DEBUG
+        settings.DEBUG = True
+        query_count = len(connection.queries)
+        func(*args, **kwargs)
+        self.failUnlessEqual(num, len(connection.queries) - query_count)
+        connection.queries = connection.queries[:query_count]
+        settings.DEBUG = old_debug
+    
+    def test_bulk(self):
+        from django.db.models.sql.constants import GET_ITERATOR_CHUNK_SIZE
+        s = S.objects.create(r=R.objects.create())
+        for i in xrange(2*GET_ITERATOR_CHUNK_SIZE):
+            T.objects.create(s=s)
+        #   1 (select related `T` instances)
+        # + 1 (select related `U` instances)
+        # + 2 (delete `T` instances in batches)
+        # + 1 (delete `s`)
+        self.assert_num_queries(5, s.delete)
+        self.failIf(S.objects.exists())
+        
+    def test_instance_update(self):
+        deleted = []
+        related_setnull_sets = []
+        def pre_delete(sender, **kwargs):
+            obj = kwargs['instance']
+            deleted.append(obj)
+            if isinstance(obj, R):
+                related_setnull_sets.append(list(a.pk for a in obj.setnull_set.all()))
+
+        models.signals.pre_delete.connect(pre_delete)
+        a = create_a('update_setnull')
+        a.setnull.delete()
+        
+        a = create_a('update_cascade')
+        a.cascade.delete()
+        
+        for obj in deleted:
+            self.failUnlessEqual(None, obj.pk)
+            
+        for pk_list in related_setnull_sets:
+            for a in A.objects.filter(id__in=pk_list):
+                self.failUnlessEqual(None, a.setnull)
+        
+        models.signals.pre_delete.disconnect(pre_delete)
+
+    def test_deletion_order(self):
+        pre_delete_order = []
+        post_delete_order = []
+
+        def log_post_delete(sender, **kwargs):
+            pre_delete_order.append((sender, kwargs['instance'].pk))
+
+        def log_pre_delete(sender, **kwargs):
+            post_delete_order.append((sender, kwargs['instance'].pk))
+        
+        models.signals.post_delete.connect(log_post_delete)
+        models.signals.pre_delete.connect(log_pre_delete)
+        
+        r = R.objects.create(pk=1)
+        s1 = S.objects.create(pk=1, r=r)
+        s2 = S.objects.create(pk=2, r=r)
+        t1 = T.objects.create(pk=1, s=s1)
+        t2 = T.objects.create(pk=2, s=s2)
+        r.delete()
+        self.failUnlessEqual(pre_delete_order, [(T, 2), (T, 1), (S, 2), (S, 1), (R, 1)])
+        self.failUnlessEqual(post_delete_order, [(T, 1), (T, 2), (S, 1), (S, 2), (R, 1)])
+        
+        models.signals.post_delete.disconnect(log_post_delete)
+        models.signals.post_delete.disconnect(log_pre_delete)
+        
Index: tests/regressiontests/admin_util/tests.py
===================================================================
--- tests/regressiontests/admin_util/tests.py	(revision 14291)
+++ tests/regressiontests/admin_util/tests.py	(working copy)
@@ -26,46 +26,37 @@
     def _check(self, target):
         self.assertEquals(self.n.nested(lambda obj: obj.num), target)
 
-    def _add(self, obj, parent=None):
-        # don't bother providing the extra args that NestedObjects ignores
-        self.n.add(None, None, obj, None, parent)
+    def _connect(self, i, j):
+        self.objs[i].parent = self.objs[j]
+        self.objs[i].save()
+        
+    def _collect(self, *indices):
+        self.n.collect([self.objs[i] for i in indices])
 
     def test_unrelated_roots(self):
-        self._add(self.objs[0])
-        self._add(self.objs[1])
-        self._add(self.objs[2], self.objs[1])
-
+        self._connect(2, 1)
+        self._collect(0)
+        self._collect(1)
         self._check([0, 1, [2]])
 
     def test_siblings(self):
-        self._add(self.objs[0])
-        self._add(self.objs[1], self.objs[0])
-        self._add(self.objs[2], self.objs[0])
-
+        self._connect(1, 0)
+        self._connect(2, 0)
+        self._collect(0)
         self._check([0, [1, 2]])
 
-    def test_duplicate_instances(self):
-        self._add(self.objs[0])
-        self._add(self.objs[1])
-        dupe = Count.objects.get(num=1)
-        self._add(dupe, self.objs[0])
-
-        self._check([0, 1])
-
     def test_non_added_parent(self):
-        self._add(self.objs[0], self.objs[1])
-
+        self._connect(0, 1)
+        self._collect(0)
         self._check([0])
 
     def test_cyclic(self):
-        self._add(self.objs[0], self.objs[2])
-        self._add(self.objs[1], self.objs[0])
-        self._add(self.objs[2], self.objs[1])
-        self._add(self.objs[0], self.objs[2])
-
+        self._connect(0, 2)
+        self._connect(1, 0)
+        self._connect(2, 1)
+        self._collect(0)
         self._check([0, [1, [2]]])
 
-
 class UtilTests(unittest.TestCase):
     def test_values_from_lookup_field(self):
         """
Index: tests/regressiontests/admin_util/models.py
===================================================================
--- tests/regressiontests/admin_util/models.py	(revision 14291)
+++ tests/regressiontests/admin_util/models.py	(working copy)
@@ -18,7 +18,11 @@
 
 class Count(models.Model):
     num = models.PositiveSmallIntegerField()
+    parent = models.ForeignKey('self', null=True)
 
+    def __unicode__(self):
+        return unicode(self.num)
+
 class Event(models.Model):
     date = models.DateTimeField(auto_now_add=True)
 
