PB-1575: make deletion batch size configurable.

One `delete()` per object is very slow. One `delete()` for all the objects
consumes a lot of memory and fails in an all-or-nothing mode (i.e. when it
fails mid-way, nothing gets deleted).

With this change, we split the objects into batches of configurable size and
run `delete()` over each batch. It consumes a ~constant amount of memory, is
faster than having a batch size of 1 and is more robust. I.e. when it fails
mid-way, it will most likely have deleted at least some objects and the next
run has that much less work to do.

There will be cases where expired items are partially deleted (i.e. some of
their assets are deleted but not all). I don't expect that to be a problem for
our use case.

Author: adk-swisstopo

app/config/settings_prod.py

@@ -209,6 +209,7 @@
 DELETE_EXPIRED_ITEMS_OLDER_THAN_HOURS = 24
 DELETE_EXPIRED_ITEMS_MAX = 110 * 1000
 DELETE_EXPIRED_ITEMS_MAX_PERCENTAGE = 50
+DELETE_EXPIRED_ITEMS_BATCH_SIZE = 10 * 1000
 
 # Media files (i.e. uploaded content=assets in this project)
 UPLOAD_FILE_CHUNK_SIZE = 1024 * 1024  # Size in Bytes

app/stac_api/management/commands/remove_expired_items.py

@@ -3,6 +3,7 @@
 
 from django.conf import settings
 from django.core.management.base import CommandParser
+from django.core.paginator import Paginator
 from django.utils import timezone
 
 from stac_api.models.general import BaseAssetUpload
@@ -22,11 +23,53 @@ def __str__(self):
 
 class Handler(CommandHandler):
 
-    def delete(self, instance, object_type):
-        if self.options['dry_run']:
-            self.print_success(f'skipping deletion of {object_type} {instance}')
-        else:
-            instance.delete()
+    def delete_by_batch(self, queryset, object_type, batch_size):
+        # When many rows are involved, looping over each one is very slow.
+        # Running a single delete() against all of them consumes a lot of memory
+        # and does not delete anything if it fails mid-way. Hence, we batch.
+        #
+        # Django's delete() method already batches deletions in groups of 100
+        # rows. These batches are wrapped within transactions. It does not seem
+        # to be designed to allow disabling the transaction or tweaking the
+        # batch size.
+        # https://github.com/django/django/blob/main/django/db/models/sql/subqueries.py#L26
+        # https://github.com/django/django/blob/main/django/db/models/deletion.py#L454
+        # Also, it does not seem to do anything to reduce memory consumption.
+        #
+        # In our case, we don't need the deletions to be transactional. If we
+        # die in the middle, it's fine if some rows are deleted and some are
+        # not. We can remove the remaining rows next time we run. That's better
+        # than waiting forever, to fail and to have to start from scratch next
+        # time.
+        type_name = f'stac_api.{object_type.__name__}'
+        deleted_count = 0
+        # We delete rows as we iterate over them. This only works if we iterate
+        # from the end to the beginning. But we also want to delete the objects
+        # in the order of the QuerySet we received. Hence, we first reverse the
+        # the QuerySet then we reverse the iterator.
+        queryset = queryset.reverse()
+        paginator = Paginator(queryset, batch_size)
+        for page_number in reversed(paginator.page_range):
+            page = paginator.page(page_number)
+            # We cannot just call page.object_list.delete() because DELETE
+            # does not support LIMIT/OFFSET. So instead we extract the ids
+            # then we'll build a new QuerySet to DELETE them.
+            ids = page.object_list.values('id')
+            expected_deletions = len(ids)
+            dry_run_prefix = ''
+            if self.options['dry_run']:
+                dry_run_prefix = '[dry run]: '
+                deleted_objs = {}
+                actual_deletions = expected_deletions
+            else:
+                (_, deleted_objs) = object_type.objects.filter(id__in=ids).delete()
+                actual_deletions = deleted_objs.get(type_name, 0)
+            deleted_count += actual_deletions
+            self.print_success(
+                f'{dry_run_prefix}Deleted {deleted_count}/{paginator.count} {type_name}.'
+                f' In this batch: {actual_deletions}/{expected_deletions}.'
+                f' All objects in this batch: {deleted_objs}.'
+            )
 
     def _raise_if_too_many_deletions(self, max_deletions, max_deletions_pct, items_count):
         if items_count > max_deletions:
@@ -42,18 +85,20 @@ def _raise_if_too_many_deletions(self, max_deletions, max_deletions_pct, items_c
 
     def run(self):
         self.print_success('running command to remove expired items')
+        batch_size = self.options['batch_size']
         min_age_hours = self.options['min_age_hours']
         max_deletions = self.options['max_deletions']
         max_deletions_pct = self.options['max_deletions_percentage']
         self.print_warning(
             f"deleting no more than {max_deletions} or "
             f"{max_deletions_pct}%% items expired for longer"
-            f" than {min_age_hours} hours"
+            f" than {min_age_hours} hours, {batch_size} at a time"
         )
 
         expiration = timezone.now() - timedelta(hours=min_age_hours)
 
-        items = Item.objects.filter(properties_expires__lte=expiration)
+        items = Item.objects.filter(properties_expires__lte=expiration
+                                   ).order_by('properties_expires')
         items_count = items.count()
 
         self._raise_if_too_many_deletions(max_deletions, max_deletions_pct, items_count)
@@ -69,8 +114,8 @@ def run(self):
                 "WARNING: There were still pending asset uploads for expired items. "
                 "These were likely stale, so we aborted them"
             )
-        self.delete(assets, 'assets')
-        self.delete(items, 'items')
+        self.delete_by_batch(assets, Asset, batch_size)
+        self.delete_by_batch(items, Item, batch_size)
 
         if self.options['dry_run']:
             self.print_success(f'[dry run] would have removed {items_count} expired items')
@@ -105,6 +150,13 @@ def add_arguments(self, parser: CommandParser) -> None:
             action='store_true',
             help='Simulate deleting items, without actually deleting them'
         )
+        default_batch_size = settings.DELETE_EXPIRED_ITEMS_BATCH_SIZE
+        parser.add_argument(
+            '--batch-size',
+            type='positive_int',
+            default=default_batch_size,
+            help=f"How many rows to delete at a time ({default_batch_size})"
+        )
         default_min_age = settings.DELETE_EXPIRED_ITEMS_OLDER_THAN_HOURS
         parser.add_argument(
             '--min-age-hours',

app/tests/tests_10/test_remove_expired_items.py

@@ -230,11 +230,6 @@ def test_remove_item_dry_run(self):
             expected_stdout_patterns=[
                 "running command to remove expired items",
                 "deleting no more than 110000 or 50% items expired for longer than 24 hours",
-                (
-                    "skipping deletion of assets <QuerySet"
-                    " [<Asset: asset-0.tiff>, <Asset: asset-1.tiff>]>"
-                ),
-                "skipping deletion of items <ItemQuerySet [<Item: collection-1/item-0>]>",
                 "[dry run] would have removed 1 expired items",
             ]
         )
@@ -331,6 +326,12 @@ class RemoveExpiredItemsManyWithProfiling(RemoveExpiredItemsBase):
     def test_remove_item(self):
         self.run_test(command_args=["--max-deletions-percentage=99"], expected_stdout_patterns=[])
 
+    def test_multiple_batches(self):
+        self.run_test(
+            command_args=["--max-deletions-percentage=99", "--batch-size=7"],
+            expected_stdout_patterns=[]
+        )
+
     @staticmethod
     def _diff_memory(before, after):
         diff = after.compare_to(before, 'filename')