Skip to content

fix: avoid pushing all links of record at once #7506

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
merged 1 commit into from
Jan 27, 2024
Merged

fix: avoid pushing all links of record at once #7506

Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
70 changes: 35 additions & 35 deletions packages/nocodb/src/modules/jobs/jobs/at-import/helpers/readAndProcessData.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -17,7 +17,7 @@ const BULK_DATA_BATCH_COUNT =
const BULK_DATA_BATCH_SIZE =
+process.env.AT_IMPORT_BULK_DATA_BATCH_SIZE || 20 * 1024; // import N bytes at a time
const BULK_LINK_BATCH_COUNT =
+process.env.AT_IMPORT_BULK_LINK_BATCH_COUNT || 200; // import N links at a time
+process.env.AT_IMPORT_BULK_LINK_BATCH_COUNT || 500; // import N links at a time
const BULK_PARALLEL_PROCESS = +process.env.AT_IMPORT_BULK_PARALLEL_PROCESS || 2; // process N records at a time
const STREAM_BUFFER_LIMIT = +process.env.AT_IMPORT_STREAM_BUFFER_LIMIT || 100; // pause reading if we have more than N records to avoid backpressure
const QUEUE_BUFFER_LIMIT = +process.env.AT_IMPORT_QUEUE_BUFFER_LIMIT || 20; // pause streaming if we have more than N records in the queue
Expand Down Expand Up @@ -406,44 +406,44 @@ export async function importLTARData({
const { id: _atId, ...rec } = record;

// todo: use actual alias instead of sanitized
assocTableData[assocMeta.modelMeta.id].push(
...(
rec?.[atNcAliasRef[table.id][assocMeta.colMeta.title]] || []
).map((id) => ({
const links =
rec?.[atNcAliasRef[table.id][assocMeta.colMeta.title]] || [];
for (const id of links) {
assocTableData[assocMeta.modelMeta.id].push({
[assocMeta.curCol.title]: record.id,
[assocMeta.refCol.title]: id,
})),
);

if (
assocTableData[assocMeta.modelMeta.id].length >=
BULK_LINK_BATCH_COUNT
) {
let insertArray = assocTableData[
assocMeta.modelMeta.id
].splice(0, assocTableData[assocMeta.modelMeta.id].length);

const lastImportedCount = importedCount;
importedCount += insertArray.length;

logBasic(
`:: Importing '${
table.title
}' LTAR data :: ${lastImportedCount} - ${
lastImportedCount + insertArray.length
}`,
);

await services.bulkDataService.bulkDataInsert({
baseName,
tableName: assocMeta.modelMeta.id,
body: insertArray,
cookie: {},
skip_hooks: true,
foreign_key_checks: !!source.isMeta(),
});

insertArray = [];
if (
assocTableData[assocMeta.modelMeta.id].length >=
BULK_LINK_BATCH_COUNT
) {
let insertArray = assocTableData[
assocMeta.modelMeta.id
].splice(0, assocTableData[assocMeta.modelMeta.id].length);

const lastImportedCount = importedCount;
importedCount += insertArray.length;

logBasic(
`:: Importing '${
table.title
}' LTAR data :: ${lastImportedCount} - ${
lastImportedCount + insertArray.length
}`,
);

await services.bulkDataService.bulkDataInsert({
baseName,
tableName: assocMeta.modelMeta.id,
body: insertArray,
cookie: {},
skip_hooks: true,
foreign_key_checks: !!source.isMeta(),
});

insertArray = [];
}
}

if (queue.size < QUEUE_BUFFER_LIMIT / 2) dataStream.resume();
Expand Down