A work.migrated page reads only its own issues' events
Each page rewrote the log in four statements that each read every event of the page's repositories, once per page: a large repository took longer than D1 allows, and the page, failing, was retried for ever without the catch-up's cursor moving. Migration 0015 indexes the events the rewrite looks for by repository and old number (issue events and comments, issue, otherIssue), and each statement now starts from the page's own items and finds their events through those indexes.
- G1T-Session:
- ses_01m56w5m6cegebbtcjh7ax0a5g
2 files+71−160/2 viewed
| 1 | + | -- Rewriting the log for a page of `work.migrated` (src/work_items.rs | |
| 2 | + | -- `MOVED`) finds each of the page's issues' events by its repository and | |
| 3 | + | -- old number, instead of reading every event of the page's repositories | |
| 4 | + | -- once per statement: a large repository's log took longer than D1 allows, | |
| 5 | + | -- and the page was retried for ever. Each index holds only the events the | |
| 6 | + | -- rewrite looks for, which events written since no longer are (they name | |
| 7 | + | -- work items), but for comments. | |
| 8 | + | CREATE INDEX IF NOT EXISTS events_issue_number ON events (repo_id, json_extract(data, '$.number')) | |
| 9 | + | WHERE type IN ('issue.opened', 'issue.updated', 'issue.assigned', 'issue.labeled', 'issue.unlabeled', 'issue.milestoned', | |
| 10 | + | 'issue.demilestoned', 'issue.closed', 'issue.reopened', 'comment.created', 'comment.edited', 'comment.deleted'); | |
| 11 | + | CREATE INDEX IF NOT EXISTS events_issue ON events (repo_id, json_extract(data, '$.issue')) | |
| 12 | + | WHERE json_extract(data, '$.issue') IS NOT NULL; | |
| 13 | + | CREATE INDEX IF NOT EXISTS events_other_issue ON events (repo_id, json_extract(data, '$.otherIssue')) | |
| 14 | + | WHERE json_extract(data, '$.otherIssue') IS NOT NULL; |
| 33 | 33 | "issue.reopened", | |
| 34 | 34 | ]; | |
| 35 | 35 | ||
| 36 | + | /// The page's (`?1`) issues' events and comments, by repository and old | |
| 37 | + | /// number: the `type` list is `events_issue_number`'s (migration 0015), | |
| 38 | + | /// word for word, so SQLite reads that index. | |
| 39 | + | macro_rules! page_events_by_number { | |
| 40 | + | () => { | |
| 41 | + | "SELECT e.id FROM work_migrated m JOIN events e INDEXED BY events_issue_number | |
| 42 | + | ON e.repo_id = m.repo_id AND json_extract(e.data, '$.number') = m.legacy_number | |
| 43 | + | AND e.type IN ('issue.opened', 'issue.updated', 'issue.assigned', 'issue.labeled', 'issue.unlabeled', 'issue.milestoned', | |
| 44 | + | 'issue.demilestoned', 'issue.closed', 'issue.reopened', 'comment.created', 'comment.edited', 'comment.deleted') | |
| 45 | + | WHERE m.event_id = ?1" | |
| 46 | + | }; | |
| 47 | + | } | |
| 48 | + | ||
| 36 | 49 | /// What one page does, in order: `?1` the event's id, `?2` its items as | |
| 37 | 50 | /// JSON, `?3` [`ISSUE_EVENTS`] as JSON. | |
| 38 | 51 | pub const MOVED: [&str; 12] = [ | |
| ⋯ | |||
| 73 | 86 | "UPDATE inbox_subscriptions SET thread = 'item:' || m.item_id | |
| 74 | 87 | FROM work_migrated m | |
| 75 | 88 | WHERE m.event_id = ?1 AND inbox_subscriptions.thread = m.repo_id || '#' || m.legacy_number", | |
| 76 | − | // The log. Each statement reads the page's repositories' events once, | |
| 77 | − | // and finds each one's work item by its repository and old number. | |
| 78 | − | "UPDATE events SET type = 'work_item.' || substr(type, 7), | |
| 89 | + | // The log: each of the page's issues' events, found by its repository | |
| 90 | + | // and old number through an index of only such events (migration | |
| 91 | + | // 0015), so a page reads what it rewrites and no more of the log. | |
| 92 | + | concat!( | |
| 93 | + | "UPDATE events SET type = 'work_item.' || substr(type, 7), | |
| 79 | 94 | data = json_set(json_remove(data, '$.issueId', '$.repoId'), | |
| 80 | 95 | '$.itemId', (SELECT m.item_id FROM work_migrated m WHERE m.repo_id = events.repo_id AND m.legacy_number = json_extract(events.data, '$.number')), | |
| 81 | 96 | '$.workspace', (SELECT m.workspace FROM work_migrated m WHERE m.repo_id = events.repo_id AND m.legacy_number = json_extract(events.data, '$.number')), | |
| 82 | 97 | '$.repoIds', json_array(events.repo_id), | |
| 83 | 98 | '$.number', (SELECT m.number FROM work_migrated m WHERE m.repo_id = events.repo_id AND m.legacy_number = json_extract(events.data, '$.number'))) | |
| 84 | − | WHERE repo_id IN (SELECT repo_id FROM work_migrated WHERE event_id = ?1) | |
| 85 | − | AND type IN (SELECT value FROM json_each(?3)) | |
| 86 | − | AND EXISTS (SELECT 1 FROM work_migrated m WHERE m.repo_id = events.repo_id AND m.legacy_number = json_extract(events.data, '$.number'))", | |
| 99 | + | WHERE id IN (", | |
| 100 | + | page_events_by_number!(), | |
| 101 | + | ") | |
| 102 | + | AND type IN (SELECT value FROM json_each(?3))" | |
| 103 | + | ), | |
| 87 | 104 | // Comments on an issue: on no pull request, and naming no work item yet. | |
| 88 | − | "UPDATE events SET | |
| 105 | + | concat!( | |
| 106 | + | "UPDATE events SET | |
| 89 | 107 | data = json_set(data, | |
| 90 | 108 | '$.itemId', (SELECT m.item_id FROM work_migrated m WHERE m.repo_id = events.repo_id AND m.legacy_number = json_extract(events.data, '$.number')), | |
| 91 | 109 | '$.workspace', (SELECT m.workspace FROM work_migrated m WHERE m.repo_id = events.repo_id AND m.legacy_number = json_extract(events.data, '$.number')), | |
| 92 | 110 | '$.repoIds', json_array(events.repo_id), | |
| 93 | 111 | '$.number', (SELECT m.number FROM work_migrated m WHERE m.repo_id = events.repo_id AND m.legacy_number = json_extract(events.data, '$.number'))) | |
| 94 | − | WHERE repo_id IN (SELECT repo_id FROM work_migrated WHERE event_id = ?1) | |
| 112 | + | WHERE id IN (", | |
| 113 | + | page_events_by_number!(), | |
| 114 | + | ") | |
| 95 | 115 | AND type IN ('comment.created', 'comment.edited', 'comment.deleted') | |
| 96 | − | AND json_extract(data, '$.pullId') IS NULL AND json_extract(data, '$.itemId') IS NULL | |
| 97 | − | AND EXISTS (SELECT 1 FROM work_migrated m WHERE m.repo_id = events.repo_id AND m.legacy_number = json_extract(events.data, '$.number'))", | |
| 116 | + | AND json_extract(data, '$.pullId') IS NULL AND json_extract(data, '$.itemId') IS NULL" | |
| 117 | + | ), | |
| 98 | 118 | // What names an issue it is about, or overlaps with: `issue`, `otherIssue`. | |
| 99 | 119 | "UPDATE events SET | |
| 100 | 120 | data = json_set(json_remove(data, '$.issue'), | |
| 101 | 121 | '$.item', (SELECT m.number FROM work_migrated m WHERE m.repo_id = events.repo_id AND m.legacy_number = json_extract(events.data, '$.issue'))) | |
| 102 | − | WHERE repo_id IN (SELECT repo_id FROM work_migrated WHERE event_id = ?1) | |
| 103 | − | AND json_extract(data, '$.issue') IS NOT NULL | |
| 104 | − | AND EXISTS (SELECT 1 FROM work_migrated m WHERE m.repo_id = events.repo_id AND m.legacy_number = json_extract(events.data, '$.issue'))", | |
| 122 | + | WHERE id IN ( | |
| 123 | + | SELECT e.id FROM work_migrated m JOIN events e INDEXED BY events_issue | |
| 124 | + | ON e.repo_id = m.repo_id AND json_extract(e.data, '$.issue') = m.legacy_number AND json_extract(e.data, '$.issue') IS NOT NULL | |
| 125 | + | WHERE m.event_id = ?1)", | |
| 105 | 126 | "UPDATE events SET | |
| 106 | 127 | data = json_set(json_remove(data, '$.otherIssue'), | |
| 107 | 128 | '$.otherItem', (SELECT m.number FROM work_migrated m WHERE m.repo_id = events.repo_id AND m.legacy_number = json_extract(events.data, '$.otherIssue'))) | |
| 108 | − | WHERE repo_id IN (SELECT repo_id FROM work_migrated WHERE event_id = ?1) | |
| 109 | − | AND json_extract(data, '$.otherIssue') IS NOT NULL | |
| 110 | − | AND EXISTS (SELECT 1 FROM work_migrated m WHERE m.repo_id = events.repo_id AND m.legacy_number = json_extract(events.data, '$.otherIssue'))", | |
| 129 | + | WHERE id IN ( | |
| 130 | + | SELECT e.id FROM work_migrated m JOIN events e INDEXED BY events_other_issue | |
| 131 | + | ON e.repo_id = m.repo_id AND json_extract(e.data, '$.otherIssue') = m.legacy_number AND json_extract(e.data, '$.otherIssue') IS NOT NULL | |
| 132 | + | WHERE m.event_id = ?1)", | |
| 111 | 133 | ]; | |
| 112 | 134 | ||
| 113 | 135 | /// How many of `?1`, `?2`, `?3` a statement reads: D1 takes no more values | |
| ⋯ | |||
| 208 | 230 | } | |
| 209 | 231 | ||
| 210 | 232 | #[test] | |
| 233 | + | fn a_page_reads_only_its_own_issues_events() { | |
| 234 | + | let db = database_before("9999"); | |
| 235 | + | for sql in MOVED.iter().filter(|sql| sql.contains("UPDATE events")) { | |
| 236 | + | let mut statement = db.prepare(&format!("EXPLAIN QUERY PLAN {sql}")).unwrap(); | |
| 237 | + | let plan: Vec<String> = statement | |
| 238 | + | .query_map(rusqlite::params_from_iter(["evt_9", "[]", "[]"].iter().take(bound(sql))), |row| row.get::<_, String>(3)) | |
| 239 | + | .unwrap() | |
| 240 | + | .map(Result::unwrap) | |
| 241 | + | .collect(); | |
| 242 | + | // Never the whole log, nor every event of the page's repositories. | |
| 243 | + | assert!( | |
| 244 | + | plan.iter().all(|step| !step.starts_with("SCAN events") && !step.starts_with("SCAN e") && !step.contains("INDEX events_repo")), | |
| 245 | + | "{sql}\n{plan:#?}" | |
| 246 | + | ); | |
| 247 | + | assert!(plan.iter().any(|step| step.contains("INDEX events_issue") || step.contains("INDEX events_other_issue")), "{sql}\n{plan:#?}"); | |
| 248 | + | } | |
| 249 | + | } | |
| 250 | + | ||
| 251 | + | #[test] | |
| 211 | 252 | fn what_was_kept_by_issue_number_follows_the_work_item() { | |
| 212 | 253 | let db = database_before("0013_work_items.sql"); | |
| 213 | 254 | // ana: a thread on issue 3, and one on pull request 4. | |
Comments
Sign in to comment.