Worker activities statistics endpoint

added 15 commits

ed0dd197...85f21724 - 14 commits from branch master
ad77d2e4 - Base activities endpoint

Compare with previous version

added 8 commits

ad77d2e4...9392dd5b - 4 commits from branch master
203370b8 - Base activities endpoint
93f16ec8 - Use a serializer
c3c1653e - Add total elements count
437c0292 - Add tests

Compare with previous version

marked this merge request as ready

changed the description

requested review from @babadie

added 20 commits

437c0292...ccb8decb - 15 commits from branch master
7c723101 - Base activities endpoint
5c89912c - Use a serializer
a68a3b85 - Add total elements count
805709bb - Add tests
fc1e23dd - Suggestion

Compare with previous version

added 1 commit

a20626e8 - Update tests as the stats query is counted

Compare with previous version

@babadie I updated the code following your suggestion (no raw SQL), it seems much better. It takes approx. 2.5s for 4 693 915 existing WorkerActivity on my setup, for a single corpus and version.

I'll have a look on retrieving all activities for a corpus tomorrow morning

Retrieving all the activities on elements for a corpus is possible and easy to do in raw SQL. Request execution time seems linear to the total number of worker activity in my case (approx 1.5s + 0.2s/million, a bit slower when states are balanced).

However, it will be hard to implement using django ORM as it requires a custom GROUP BY.

SELECT 
  "dataimport_workeractivity"."worker_version_id",
  (COALESCE(SUM(CASE WHEN state = 'queued' THEN 1 ELSE 0 END), 0)) AS "queued",
  (COALESCE(SUM(CASE WHEN state = 'started' THEN 1 ELSE 0 END), 0)) AS "started",
  (COALESCE(SUM(CASE WHEN state = 'processed' THEN 1 ELSE 0 END), 0)) AS "processed",
  (COALESCE(SUM(CASE WHEN state = 'error' THEN 1 ELSE 0 END), 0)) AS "error"
FROM "dataimport_workeractivity"
WHERE "dataimport_workeractivity"."element_id" IN (
  SELECT U0."id"
    FROM "documents_element" U0
    WHERE U0."corpus_id" = 'cde11c80-9d98-474d-ad63-a07f11a6ad27'::uuid
)
GROUP BY ("dataimport_workeractivity"."worker_version_id")

Thanks to @erwanrouchet for this one:

WorkerActivity.objects \
    .filter(element_id__in=corpus.elements.values('id')) \
    .values('worker_version_id') \
    .annotate(
        queued=Count('id', filter=Q(state='queued'),
        error=Count('id', filter=Q(state='error'))
        […]
    )