From 1079a19cd90bbe2b6e550b8f606430f18abdf602 Mon Sep 17 00:00:00 2001 From: Hongbo Wu Date: Tue, 26 Dec 2023 20:36:43 +0800 Subject: [PATCH 1/6] execute update query in batch to avoid slow bulk action query --- packages/api/src/resolvers/article/index.ts | 9 +++- packages/api/src/services/library_item.ts | 47 +++++++++++++++++++-- 2 files changed, 51 insertions(+), 5 deletions(-) diff --git a/packages/api/src/resolvers/article/index.ts b/packages/api/src/resolvers/article/index.ts index ec7bd1e8e..5d698fc81 100644 --- a/packages/api/src/resolvers/article/index.ts +++ b/packages/api/src/resolvers/article/index.ts @@ -46,6 +46,7 @@ import { SearchErrorCode, SearchSuccess, SetBookmarkArticleError, + SetBookmarkArticleErrorCode, SetBookmarkArticleSuccess, SetFavoriteArticleError, SetFavoriteArticleErrorCode, @@ -70,6 +71,7 @@ import { findOrCreateLabels, } from '../../services/labels' import { + batchUpdateLibraryItems, createLibraryItem, findLibraryItemById, findLibraryItemByUrl, @@ -78,7 +80,6 @@ import { sortParamsToSort, updateLibraryItem, updateLibraryItemReadingProgress, - updateLibraryItems, } from '../../services/library_item' import { parsedContentToLibraryItem } from '../../services/save_page' import { @@ -551,6 +552,10 @@ export const setBookmarkArticleResolver = authorized< SetBookmarkArticleError, MutationSetBookmarkArticleArgs >(async (_, { input: { articleID } }, { uid, log, pubsub }) => { + if (!articleID) { + return { errorCodes: [SetBookmarkArticleErrorCode.NotFound] } + } + // delete the item and its metadata const deletedLibraryItem = await updateLibraryItem( articleID, @@ -860,7 +865,7 @@ export const bulkActionResolver = authorized< labels = await findLabelsByIds(labelIds, uid) } - await updateLibraryItems( + await batchUpdateLibraryItems( action, { query, diff --git a/packages/api/src/services/library_item.ts b/packages/api/src/services/library_item.ts index 99f143f48..e97f27ba8 100644 --- a/packages/api/src/services/library_item.ts +++ b/packages/api/src/services/library_item.ts @@ -886,7 +886,7 @@ export const countByCreatedAt = async ( ) } -export const updateLibraryItems = async ( +export const batchUpdateLibraryItems = async ( action: BulkActionType, searchArgs: SearchArgs, userId: string, @@ -902,7 +902,7 @@ export const updateLibraryItems = async ( } // build the script - let values: QueryDeepPartialEntity = {} + let values: Record = {} let addLabels = false switch (action) { case BulkActionType.Archive: @@ -984,7 +984,48 @@ export const updateLibraryItems = async ( return tx.getRepository(EntityLabel).save(labelsToAdd) } - return queryBuilder.update(LibraryItem).set(values).execute() + const countSql = queryBuilder.select('COUNT(1) INTO total_rows').getSql() + const [subQuery, params] = queryBuilder.select('id').getQueryAndParameters() + const valuesSql = Object.keys(values) + // eslint-disable-next-line @typescript-eslint/restrict-template-expressions + .map((key) => `${key} = ${values[key]}`) + .join(', ') + + const sql = ` + -- Set batch size + DO $$ + DECLARE + batch_size INT := 100; + total_rows INT := 1000; + num_batches INT; + current_offset INT; + BEGIN + -- Get the total count of rows to be updated + ${countSql}; + + -- Calculate the number of batches + num_batches := CEIL(total_rows * 1.0 / batch_size); + + -- Loop through batches + FOR i IN 0..num_batches-1 LOOP + -- Set the current offset + current_offset := i * batch_size; + + -- Perform incremental update in batches using LIMIT and OFFSET + UPDATE omnivore.library_item + SET ${valuesSql} + FROM ( + ${subQuery} + ORDER BY id + LIMIT batch_size + OFFSET current_offset + ) AS batch + WHERE library_item.id = batch.id; + END LOOP; + END $$ + ` + + return tx.query(sql, params) }) } From 55518139fb3c7417706d676bb044a1785f62d798 Mon Sep 17 00:00:00 2001 From: Hongbo Wu Date: Thu, 28 Dec 2023 16:02:39 +0800 Subject: [PATCH 2/6] generate raw sql because postgres does not support prepared statements in DO blocks --- packages/api/src/repository/index.ts | 58 ++++++++++++++++- packages/api/src/services/library_item.ts | 41 +++++++----- packages/api/test/resolvers/article.test.ts | 72 ++++++++++++++------- 3 files changed, 129 insertions(+), 42 deletions(-) diff --git a/packages/api/src/repository/index.ts b/packages/api/src/repository/index.ts index 9e1dd187d..9d6ce825a 100644 --- a/packages/api/src/repository/index.ts +++ b/packages/api/src/repository/index.ts @@ -1,5 +1,5 @@ import * as httpContext from 'express-http-context2' -import { EntityManager, EntityTarget, Repository } from 'typeorm' +import { EntityManager, EntityTarget, QueryBuilder, Repository } from 'typeorm' import { appDataSource } from '../data_source' import { Claims } from '../resolvers/types' import { SetClaimsRole } from '../utils/dictionary' @@ -45,3 +45,59 @@ export const authTrx = async ( export const getRepository = (entity: EntityTarget) => { return appDataSource.getRepository(entity) } + +export const queryBuilderToRawSql = (q: QueryBuilder): string => { + const queryAndParams = q.getQueryAndParameters() + let sql = queryAndParams[0] + const params = queryAndParams[1] + + params.forEach((value, index) => { + if (typeof value === 'string') { + sql = sql.replace(`$${index + 1}`, `'${value}'`) + } else if (typeof value === 'object') { + if (Array.isArray(value)) { + sql = sql.replace( + `$${index + 1}`, + value + .map((element) => { + if (typeof element === 'string') { + return `'${element}'` + } + + if (typeof element === 'number' || typeof element === 'boolean') { + return element.toString() + } + }) + .join(',') + ) + } else if (value instanceof Date) { + sql = sql.replace(`$${index + 1}`, `'${value.toISOString()}'`) + } + } else if (typeof value === 'number' || typeof value === 'boolean') { + sql = sql.replace(`$${index + 1}`, value.toString()) + } + }) + + return sql +} + +export const valuesToRawSql = ( + values: Record +): string => { + let sql = '' + + Object.keys(values).forEach((key, index) => { + const value = values[key] + if (typeof value === 'string') { + sql += `${key} = '${value}'` + } else { + sql += `${key} = ${value.toString()}` + } + + if (index < Object.keys(values).length - 1) { + sql += ', ' + } + }) + + return sql +} diff --git a/packages/api/src/services/library_item.ts b/packages/api/src/services/library_item.ts index e97f27ba8..95a79a663 100644 --- a/packages/api/src/services/library_item.ts +++ b/packages/api/src/services/library_item.ts @@ -8,7 +8,12 @@ import { Label } from '../entity/label' import { LibraryItem, LibraryItemState } from '../entity/library_item' import { BulkActionType, InputMaybe, SortParams } from '../generated/graphql' import { createPubSubClient, EntityType } from '../pubsub' -import { authTrx, getColumns } from '../repository' +import { + authTrx, + getColumns, + queryBuilderToRawSql, + valuesToRawSql, +} from '../repository' import { libraryItemRepository } from '../repository/library_item' import { wordsCount } from '../utils/helpers' import { parseSearchQuery } from '../utils/search' @@ -901,20 +906,21 @@ export const batchUpdateLibraryItems = async ( return 'folder' in args } + const now = new Date().toISOString() // build the script - let values: Record = {} + let values: Record = {} let addLabels = false switch (action) { case BulkActionType.Archive: values = { - archivedAt: new Date(), + archived_at: now, state: LibraryItemState.Archived, } break case BulkActionType.Delete: values = { state: LibraryItemState.Deleted, - deletedAt: new Date(), + deleted_at: now, } break case BulkActionType.AddLabels: @@ -922,9 +928,9 @@ export const batchUpdateLibraryItems = async ( break case BulkActionType.MarkAsRead: values = { - readAt: new Date(), - readingProgressTopPercent: 100, - readingProgressBottomPercent: 100, + read_at: now, + reading_progress_top_percent: 100, + reading_progress_bottom_percent: 100, } break case BulkActionType.MoveToFolder: @@ -934,7 +940,7 @@ export const batchUpdateLibraryItems = async ( values = { folder: args.folder, - savedAt: new Date(), + saved_at: now, } break @@ -984,19 +990,20 @@ export const batchUpdateLibraryItems = async ( return tx.getRepository(EntityLabel).save(labelsToAdd) } - const countSql = queryBuilder.select('COUNT(1) INTO total_rows').getSql() - const [subQuery, params] = queryBuilder.select('id').getQueryAndParameters() - const valuesSql = Object.keys(values) - // eslint-disable-next-line @typescript-eslint/restrict-template-expressions - .map((key) => `${key} = ${values[key]}`) - .join(', ') + // generate raw sql because postgres doesn't support prepared statements in DO blocks + const countSql = queryBuilderToRawSql( + queryBuilder.select('COUNT(1) INTO total_rows') + ) + const subQuery = queryBuilderToRawSql(queryBuilder.select('id')) + const valuesSql = valuesToRawSql(values) + const batchSize = 100 const sql = ` -- Set batch size DO $$ DECLARE - batch_size INT := 100; - total_rows INT := 1000; + batch_size INT := ${batchSize}; + total_rows INT; num_batches INT; current_offset INT; BEGIN @@ -1025,7 +1032,7 @@ export const batchUpdateLibraryItems = async ( END $$ ` - return tx.query(sql, params) + return tx.query(sql) }) } diff --git a/packages/api/test/resolvers/article.test.ts b/packages/api/test/resolvers/article.test.ts index f2421e38a..3bc0bf443 100644 --- a/packages/api/test/resolvers/article.test.ts +++ b/packages/api/test/resolvers/article.test.ts @@ -2107,31 +2107,31 @@ describe('Article API', () => { } ` - before(async () => { - // Create some test items - for (let i = 0; i < 5; i++) { - await createLibraryItem( - { - user, - itemType: i == 0 ? PageType.Article : PageType.File, - title: 'test item', - readableContent: '

test

', - slug: '', - state: - i == 0 ? LibraryItemState.Failed : LibraryItemState.Succeeded, - originalUrl: `https://blog.omnivore.app/p/bulk-action-${i}`, - }, - user.id - ) - } - }) - - after(async () => { - // Delete all items - await deleteLibraryItemsByUserId(user.id) - }) - context('when action is MarkAsRead and query is in:unread', () => { + before(async () => { + // Create some test items + for (let i = 0; i < 5; i++) { + await createLibraryItem( + { + user, + itemType: i == 0 ? PageType.Article : PageType.File, + title: 'test item', + readableContent: '

test

', + slug: '', + state: + i == 0 ? LibraryItemState.Failed : LibraryItemState.Succeeded, + originalUrl: `https://blog.omnivore.app/p/bulk-action-${i}`, + }, + user.id + ) + } + }) + + after(async () => { + // Delete all items + await deleteLibraryItemsByUserId(user.id) + }) + it('marks unread items as read', async () => { const res = await graphqlRequest( bulkActionQuery(BulkActionType.MarkAsRead, 'is:unread'), @@ -2199,6 +2199,30 @@ describe('Article API', () => { ) context('when action is Delete', () => { + before(async () => { + // Create some test items + for (let i = 0; i < 5; i++) { + await createLibraryItem( + { + user, + itemType: i == 0 ? PageType.Article : PageType.File, + title: 'test item', + readableContent: '

test

', + slug: '', + state: + i == 0 ? LibraryItemState.Failed : LibraryItemState.Succeeded, + originalUrl: `https://blog.omnivore.app/p/bulk-action-${i}`, + }, + user.id + ) + } + }) + + after(async () => { + // Delete all items + await deleteLibraryItemsByUserId(user.id) + }) + it('deletes all items', async () => { const res = await graphqlRequest( bulkActionQuery(BulkActionType.Delete), From de027ee93f1d665f69499bf2afdfcde94da1e6d3 Mon Sep 17 00:00:00 2001 From: Hongbo Wu Date: Thu, 28 Dec 2023 21:57:00 +0800 Subject: [PATCH 3/6] use cursor and fetch instead of limit and offset because the subquery result set could change during update --- packages/api/src/services/library_item.ts | 40 ++++++++++++++--------- 1 file changed, 25 insertions(+), 15 deletions(-) diff --git a/packages/api/src/services/library_item.ts b/packages/api/src/services/library_item.ts index 95a79a663..cbcbf444b 100644 --- a/packages/api/src/services/library_item.ts +++ b/packages/api/src/services/library_item.ts @@ -994,10 +994,12 @@ export const batchUpdateLibraryItems = async ( const countSql = queryBuilderToRawSql( queryBuilder.select('COUNT(1) INTO total_rows') ) - const subQuery = queryBuilderToRawSql(queryBuilder.select('id')) + const subQuery = queryBuilderToRawSql( + queryBuilder.select('id').orderBy('id') + ) const valuesSql = valuesToRawSql(values) - const batchSize = 100 + const batchSize = 10 const sql = ` -- Set batch size DO $$ @@ -1005,7 +1007,9 @@ export const batchUpdateLibraryItems = async ( batch_size INT := ${batchSize}; total_rows INT; num_batches INT; - current_offset INT; + batch_id UUID; + batch_cursor CURSOR FOR + ${subQuery}; BEGIN -- Get the total count of rows to be updated ${countSql}; @@ -1013,22 +1017,28 @@ export const batchUpdateLibraryItems = async ( -- Calculate the number of batches num_batches := CEIL(total_rows * 1.0 / batch_size); + -- Open a cursor + OPEN batch_cursor; + -- Loop through batches FOR i IN 0..num_batches-1 LOOP - -- Set the current offset - current_offset := i * batch_size; + -- Fetch the next batch of IDs + FOR j IN 0..batch_size-1 LOOP + -- Fetch the next ID + FETCH batch_cursor INTO batch_id; - -- Perform incremental update in batches using LIMIT and OFFSET - UPDATE omnivore.library_item - SET ${valuesSql} - FROM ( - ${subQuery} - ORDER BY id - LIMIT batch_size - OFFSET current_offset - ) AS batch - WHERE library_item.id = batch.id; + -- Exit the loop if no more rows + EXIT WHEN NOT FOUND; + + -- Perform incremental update for the current ID + UPDATE omnivore.library_item + SET ${valuesSql} + WHERE id = batch_id; + END LOOP; END LOOP; + + -- Close the cursor + CLOSE batch_cursor; END $$ ` From b5d84098d0fb642764933d7b60cc40fa48e7808c Mon Sep 17 00:00:00 2001 From: Hongbo Wu Date: Thu, 28 Dec 2023 21:58:54 +0800 Subject: [PATCH 4/6] increase batch size to 100 --- packages/api/src/services/library_item.ts | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/packages/api/src/services/library_item.ts b/packages/api/src/services/library_item.ts index cbcbf444b..add98f09c 100644 --- a/packages/api/src/services/library_item.ts +++ b/packages/api/src/services/library_item.ts @@ -999,7 +999,7 @@ export const batchUpdateLibraryItems = async ( ) const valuesSql = valuesToRawSql(values) - const batchSize = 10 + const batchSize = 100 const sql = ` -- Set batch size DO $$ From cbb07318e54e96d57b4e06007888572891a7e141 Mon Sep 17 00:00:00 2001 From: Hongbo Wu Date: Fri, 29 Dec 2023 12:53:49 +0800 Subject: [PATCH 5/6] get the unchanged result set of sub query by checking updated_at --- packages/api/src/services/library_item.ts | 51 ++++++----------------- 1 file changed, 13 insertions(+), 38 deletions(-) diff --git a/packages/api/src/services/library_item.ts b/packages/api/src/services/library_item.ts index add98f09c..6bca196b6 100644 --- a/packages/api/src/services/library_item.ts +++ b/packages/api/src/services/library_item.ts @@ -991,54 +991,29 @@ export const batchUpdateLibraryItems = async ( } // generate raw sql because postgres doesn't support prepared statements in DO blocks - const countSql = queryBuilderToRawSql( - queryBuilder.select('COUNT(1) INTO total_rows') - ) - const subQuery = queryBuilderToRawSql( - queryBuilder.select('id').orderBy('id') - ) + const countSql = queryBuilderToRawSql(queryBuilder.select('COUNT(1)')) + const subQuery = queryBuilderToRawSql(queryBuilder.select('id')) const valuesSql = valuesToRawSql(values) - const batchSize = 100 + const start = new Date().toISOString() + const batchSize = 1000 const sql = ` -- Set batch size DO $$ DECLARE batch_size INT := ${batchSize}; - total_rows INT; - num_batches INT; - batch_id UUID; - batch_cursor CURSOR FOR - ${subQuery}; BEGIN - -- Get the total count of rows to be updated - ${countSql}; - - -- Calculate the number of batches - num_batches := CEIL(total_rows * 1.0 / batch_size); - - -- Open a cursor - OPEN batch_cursor; - -- Loop through batches - FOR i IN 0..num_batches-1 LOOP - -- Fetch the next batch of IDs - FOR j IN 0..batch_size-1 LOOP - -- Fetch the next ID - FETCH batch_cursor INTO batch_id; - - -- Exit the loop if no more rows - EXIT WHEN NOT FOUND; - - -- Perform incremental update for the current ID - UPDATE omnivore.library_item - SET ${valuesSql} - WHERE id = batch_id; - END LOOP; + FOR i IN 0..CEIL((${countSql}) * 1.0 / batch_size) - 1 LOOP + -- Update the batch + UPDATE omnivore.library_item + SET ${valuesSql} + WHERE id = ANY( + ${subQuery} + AND updated_at < '${start}' + LIMIT batch_size + ); END LOOP; - - -- Close the cursor - CLOSE batch_cursor; END $$ ` From 7153014a87ac3526d740058aa607220bb6316f74 Mon Sep 17 00:00:00 2001 From: Hongbo Wu Date: Fri, 29 Dec 2023 16:36:55 +0800 Subject: [PATCH 6/6] fix duplicate labels --- packages/api/src/services/library_item.ts | 5 +++-- 1 file changed, 3 insertions(+), 2 deletions(-) diff --git a/packages/api/src/services/library_item.ts b/packages/api/src/services/library_item.ts index 6bca196b6..24670201e 100644 --- a/packages/api/src/services/library_item.ts +++ b/packages/api/src/services/library_item.ts @@ -979,10 +979,11 @@ export const batchUpdateLibraryItems = async ( .map((label) => ({ labelId: label.id, libraryItemId: libraryItem.id, + name: label.name, })) .filter((entityLabel) => { - const existingLabel = libraryItem.labels?.find( - (l) => l.id === entityLabel.labelId + const existingLabel = libraryItem.labelNames?.find( + (l) => l.toLowerCase() === entityLabel.name.toLowerCase() ) return !existingLabel })