[batch] Drop dead rows from `job_group_inst_coll_cancellable_resources`

[daniel-goldstein]

What happened?

See here for context on why this table exists and how it is used. Records are added or updated in this table whenever jobs are added to the database or after an attempt for a job completes. Records are currently only removed when the records belong to a cancelled job group. If a job group runs to completion, we end up with many rows in the database that no longer serve any purpose, and (if you sum over the token column), have 0s for all the job columns. This does not affect correctness, but is a lot of wasted space in the database. This leads to two points that together would save a lot of space in the database (I've not quantified how much but select count(*) on this table takes longer than I've been willing to wait.

1. Rows in this table with the same key (batch_id, update_id, job_group_id, inst_coll) but different token value can be "compacted" into one row with key (batch_id, update_id, job_group_id, inst_coll, 0) (token 0) where all the other columns are summed. This is most useful for cold rows.
2. Rows whose n_*_jobs and *_cancellable_cores_mcpu columns are 0 can be deleted.

We already do 1 for the aggregated billing tables. Use tokens for parallelism on hot rows and then compact records so that records from before the current day always end up only using 1 row.

Implementing 1 should be a big win for the size of this table. Following that up with 2 would eliminate what I presume to be the vast majority of data in this table.

Version

0.2.132

Relevant log output

No response

[ehigham]

Hi @daniel-goldstein. I've taken a look at a few of the records in this table in prod. Do you know what it means to have negative values for n_running_cancellable_jobs and/or running_cancellable_cores_mcpu? Here's an example:

select * from job_group_inst_coll_cancellable_resources where batch_id = X;
+----------+-------+--------------------------+------------------------------+----------------------------+--------------------------------+-----------+-----------------------------+-----------+--------------+
| batch_id | token | n_ready_cancellable_jobs | ready_cancellable_cores_mcpu | n_running_cancellable_jobs | running_cancellable_cores_mcpu | inst_coll | n_creating_cancellable_jobs | update_id | job_group_id |
+----------+-------+--------------------------+------------------------------+----------------------------+--------------------------------+-----------+-----------------------------+-----------+--------------+
|        X |     0 |                        0 |                            0 |                          0 |                              0 | standard  |                           0 |         1 |            0 |
|        X |    18 |                        0 |                            0 |                         -1 |                          -2000 | standard  |                           0 |         1 |            0 |
|        X |    59 |                        0 |                            0 |                          0 |                              0 | standard  |                           0 |         1 |            0 |
|        X |    60 |                        0 |                            0 |                         -1 |                          -2000 | standard  |                           0 |         1 |            0 |
|        X |    74 |                        0 |                            0 |                          0 |                              0 | standard  |                           0 |         1 |            0 |
|        X |   172 |                        0 |                            0 |                          0 |                              0 | standard  |                           0 |         1 |            0 |
|        X |   185 |                        0 |                            0 |                         -1 |                          -1000 | standard  |                           0 |         1 |            0 |
+----------+-------+--------------------------+------------------------------+----------------------------+--------------------------------+-----------+-----------------------------+-----------+--------------+

[ehigham]

WIP solution. We need to fetch resources for job groups that have completed but have not been cancelled.
The job_groups table describes if job group has been cancelled. Simply joining the job groups to job_group_inst_coll_cancellable_resources is prohibitively expensive due to explodes. My approach is to query for finished non-cancelled job groups from a last (batch, job) id then process records iteratively:

sql = """\
WITH T AS ( 
    SELECT G.batch_id, G.job_group_id
    FROM job_groups AS G 
    INNER JOIN job_group_self_and_ancestors AS D
       ON G.batch_id     = D.batch_id
      AND G.job_group_id = D.job_group_id
    LEFT JOIN job_groups_cancelled AS C 
       ON C.id           = G.batch_id
      AND C.job_group_id = D.ancestor_id 
    WHERE G.batch_id >= ?
      AND G.job_group_id > ?
      AND G.time_completed IS NOT NULL
      AND C.id IS NULL
    ORDER BY G.batch_id ASC, G.job_group_id ASC
    LIMIT 1000
)
SELECT group_resources.batch_id
     , group_resources.update_id
     , group_resources.job_group_id
     , group_resources.inst_coll
     , SUM(group_resources.n_creating_cancellable_jobs)    AS n_creating_cancellable_jobs
     , SUM(group_resources.n_ready_cancellable_jobs)       AS n_ready_cancellable_jobs
     , SUM(group_resources.n_running_cancellable_jobs)     AS n_running_cancellable_jobs
     , SUM(group_resources.ready_cancellable_cores_mcpu)   AS ready_cancellable_cores_mcpu
     , SUM(group_resources.running_cancellable_cores_mcpu) AS running_cancellable_cores_mcpu
     , COUNT(*) as `count`
FROM job_group_inst_coll_cancellable_resources AS group_resources
INNER JOIN T USING (batch_id, job_group_id)
GROUP BY group_resources.batch_id
       , group_resources.update_id
       , group_resources.job_group_id
       , group_resources.inst_coll;
"

last_batch_id = 0
last_job_group_id = -1

while True:
   rows = db.execute(sql, [last_batch_id, last_job_group_id])
   
   if not rows:
      break

   for r in rows:
      if r.count > 1:
          delete records from job_group_inst_coll_cancellable_resources where token > 0
          update record with accumulated result in r
   
   last_batch_id = r.batch_id
   last_job_group_id = r.job_group_id         

[daniel-goldstein]

Hi @daniel-goldstein. I've taken a look at a few of the records in this table in prod. Do you know what it means to have negative values for n_running_cancellable_jobs and/or running_cancellable_cores_mcpu? Here's an example:

select * from job_group_inst_coll_cancellable_resources where batch_id = X;

+----------+-------+--------------------------+------------------------------+----------------------------+--------------------------------+-----------+-----------------------------+-----------+--------------+

| batch_id | token | n_ready_cancellable_jobs | ready_cancellable_cores_mcpu | n_running_cancellable_jobs | running_cancellable_cores_mcpu | inst_coll | n_creating_cancellable_jobs | update_id | job_group_id |

+----------+-------+--------------------------+------------------------------+----------------------------+--------------------------------+-----------+-----------------------------+-----------+--------------+

|        X |     0 |                        0 |                            0 |                          0 |                              0 | standard  |                           0 |         1 |            0 |

|        X |    18 |                        0 |                            0 |                         -1 |                          -2000 | standard  |                           0 |         1 |            0 |

|        X |    59 |                        0 |                            0 |                          0 |                              0 | standard  |                           0 |         1 |            0 |

|        X |    60 |                        0 |                            0 |                         -1 |                          -2000 | standard  |                           0 |         1 |            0 |

|        X |    74 |                        0 |                            0 |                          0 |                              0 | standard  |                           0 |         1 |            0 |

|        X |   172 |                        0 |                            0 |                          0 |                              0 | standard  |                           0 |         1 |            0 |

|        X |   185 |                        0 |                            0 |                         -1 |                          -1000 | standard  |                           0 |         1 |            0 |

+----------+-------+--------------------------+------------------------------+----------------------------+--------------------------------+-----------+-----------------------------+-----------+--------------+

Good question, looks like a bug to me as these records should eventually settle to sum to 0. Are these from recent batches? Hopefully this is some bug that has already been fixed but these rows were never cleaned up. I'd also check the code around schedule_job to see if any additions to this table aren't fault tolerant.

[ehigham]

This particular example is from quite an early batch

[ehigham]

Updated WIP query:

SELECT group_resources.batch_id
     , group_resources.update_id
     , group_resources.job_group_id
     , group_resources.inst_coll
     , SUM(group_resources.n_creating_cancellable_jobs)    AS n_creating_cancellable_jobs
     , SUM(group_resources.n_ready_cancellable_jobs)       AS n_ready_cancellable_jobs
     , SUM(group_resources.n_running_cancellable_jobs)     AS n_running_cancellable_jobs
     , SUM(group_resources.ready_cancellable_cores_mcpu)   AS ready_cancellable_cores_mcpu
     , SUM(group_resources.running_cancellable_cores_mcpu) AS running_cancellable_cores_mcpu
     , COUNT(*) as `count`
FROM job_group_inst_coll_cancellable_resources AS group_resources
INNER JOIN LATERAL (
    SELECT 1
    FROM job_groups AS G 
    INNER JOIN job_group_self_and_ancestors AS D
       ON G.batch_id     = D.batch_id
      AND G.job_group_id = D.job_group_id
    LEFT JOIN job_groups_cancelled AS C 
       ON C.id           = G.batch_id
      AND C.job_group_id = D.ancestor_id 
    WHERE G.batch_id     = group_resources.batch_id
      AND G.job_group_id = group_resources.job_group_id 
      AND G.time_completed IS NOT NULL
      AND C.id IS NULL
    LIMIT 1
) AS T ON TRUE 
WHERE group_resources.batch_id >= ?
  AND ( group_resources.batch_id
      , group_resources.update_id
      , group_resources.job_group_id
      , group_resources.inst_coll
      ) > ?
GROUP BY group_resources.batch_id
       , group_resources.update_id
       , group_resources.job_group_id
       , group_resources.inst_coll
ORDER BY group_resources.batch_id ASC
       , group_resources.update_id ASC
       , group_resources.job_group_id ASC
       , group_resources.inst_coll ASC
LIMIT 1000
;

The following is slightly faster and has the advantage of only returning results which accumulate more than one row, I'm just unsure that the ordering is correct:

SELECT R.*
FROM job_groups AS G 
INNER JOIN job_group_self_and_ancestors AS D
   ON G.batch_id     = D.batch_id
  AND G.job_group_id = D.job_group_id
LEFT JOIN job_groups_cancelled AS C 
   ON C.id           = G.batch_id
  AND C.job_group_id = D.ancestor_id 
INNER JOIN LATERAL (
    SELECT R.batch_id
         , R.update_id
         , R.job_group_id
         , R.inst_coll
         , SUM(R.n_creating_cancellable_jobs)    AS n_creating_cancellable_jobs
         , SUM(R.n_ready_cancellable_jobs)       AS n_ready_cancellable_jobs
         , SUM(R.n_running_cancellable_jobs)     AS n_running_cancellable_jobs
         , SUM(R.ready_cancellable_cores_mcpu)   AS ready_cancellable_cores_mcpu
         , SUM(R.running_cancellable_cores_mcpu) AS running_cancellable_cores_mcpu
         , COUNT(*) as `count`
    FROM job_group_inst_coll_cancellable_resources AS R
    WHERE R.batch_id     = G.batch_id
      AND R.job_group_id = G.job_group_id
    GROUP BY R.batch_id
           , R.update_id
           , R.job_group_id
           , R.inst_coll
    ORDER BY R.batch_id ASC
           , R.update_id ASC    
           , R.job_group_id ASC
           , R.inst_coll ASC
) AS R ON TRUE
WHERE G.batch_id >= ?
  AND G.time_completed IS NOT NULL
  AND C.id IS NULL
  AND ( R.batch_id
      , R.update_id
      , R.job_group_id
      , R.inst_coll
      ) > ?
  AND R.`count` > 1
LIMIT 1000;

[ehigham]

@daniel-goldstein - do you know if compacting or deleting will impact other backgroud processes? Is this table used for anything else after a job group completes?

Also, should we also compact failed job groups?