fix: replace bulk upserts with inserts/updates if needed #103
Conversation
amateima
force-pushed
the
amatei/acx-3323-replace-bulk-upserts-with-updates-only-if-necessary
branch
from
d4c340c to
c0c0348
Compare
| ); | ||
| const dbEntity = await this.postgres | ||
| .getRepository(entity) | ||
| .findOne({ where }); |
There was a problem hiding this comment.
why search a list of unique keys? isnt there a way to identify this with a single id?
There was a problem hiding this comment.
most of the events are identifiable by a pair of fields. For example, in case of a re-org, the only constant in a deposit event are depositId and originChainId fields, so you need query by these 2 fields
| } | ||
|
|
||
| // Check if the any of values of the comparison keys have changed | ||
| const isChanged = comparisonKeys.some((key) => data[key] !== dbEntity[key]); |
There was a problem hiding this comment.
how do you decide what props to compare?
There was a problem hiding this comment.
This logic will be specific for each event individually. Basically for each event you need to identify the fields that are unique in case of re-orgs and the fields that will be changed and needs to be compared
| return { | ||
| data: (await this.postgres | ||
| .getRepository(entity) | ||
| .findOne({ where })) as Entity, |
There was a problem hiding this comment.
why requery this? shouldnt it be the same as "data"
There was a problem hiding this comment.
I was thinking for this function to return the db entity because it contains the generated fields too, such as the incremental id
| .findOne({ where }); | ||
|
|
||
| if (!dbEntity) { | ||
| await this.postgres.getRepository(entity).insert(data); |
There was a problem hiding this comment.
nit: should we save the repository in a variable to use it in the lines where we are doing this.postgres.getRepository(entity)?
| }; | ||
| } | ||
|
|
||
| // Check if the any of values of the comparison keys have changed |
There was a problem hiding this comment.
| // Check if the any of values of the comparison keys have changed | |
| // Check if any of the values of the comparison keys have changed |
| data: (await this.postgres | ||
| .getRepository(entity) | ||
| .findOne({ where })) as Entity, |
There was a problem hiding this comment.
Is there a way to use the return value of update on line 89 instead of querying the entity again?
| storedEvents.deposits, | ||
| SaveQueryResultType.Inserted, | ||
| ); | ||
| await this.updateNewDepositsWithIntegratorId(newInsertedDeposits); |
There was a problem hiding this comment.
OOC: Are we ok to do this sync always?
There was a problem hiding this comment.
I wanted to give it a try again and see how much time it adds in production if we do this sync
| @@ -42,14 +48,14 @@ export class SpokePoolRepository extends dbUtils.BaseRepository { | |||
| finalised: event.blockNumber <= lastFinalisedBlock, | |||
| }; | |||
| }); | |||
| const chunkedEvents = across.utils.chunk(formattedEvents, this.chunkSize); | |||
| const chunkedEvents = across.utils.chunk(formattedEvents, 100); | |||
There was a problem hiding this comment.
should we keep this.chunkSize here?
There was a problem hiding this comment.
Done, but I lowered the default value of the chunkSize
| ["depositId", "originChainId", "transactionHash", "logIndex"], | ||
| lastFinalisedBlock, | ||
| ["transactionHash"], |
There was a problem hiding this comment.
Should we use a different set for uniqueKeys here to avoid having transactionHash both in uniqueKeys and comparisonKeys?
There was a problem hiding this comment.
This is intended here because unfortunately there's no way to identify uniquely a speedup event in case of re-orgs :(
There was a problem hiding this comment.
The outcome of this is that a re-org event would be inserted as a new one
Co-authored-by: Melisa Guevara <[email protected]>
…if-necessary' of amateima.github.com:across-protocol/indexer into amatei/acx-3323-replace-bulk-upserts-with-updates-only-if-necessary
uniqueKeysandcomparisonKeys. The result of this save operation can be (1) Nothing if the data is identical to what's stored in the database , (2) Inserted, if the unique keys are not present in the DB, (3) Finalised, if the finalised field was the only one that changed, (4) Updated, if any of the data fields were changed, (5) UpdatedAndFinalised, if both the finalised field and other fields were changedSaveQueryResult[]and filters the data based on aSaveQueryResultType, so for example, we can know which events were inserted in the DB for the first timeSpokePoolProcessorfunctions were updated to perform operations only when necessary (mostly onInsertedandUpdatedentities)