simplification
This commit is contained in:
@@ -7,20 +7,13 @@ import { sparqlEscapeUri } from '../db.js';
|
||||
|
||||
const SUBJECT_CHUNK_SIZE = Math.max(1, Number(process.env.SUBJECT_CHUNK_SIZE) || 100);
|
||||
|
||||
export async function fetchColumns(sessionQuery, subjects, spec, values = new Map(), onProgress = null) {
|
||||
export async function fetchColumns(sessionQuery, subjects, spec, values = new Map()) {
|
||||
const groups = groupColumns(spec.columns);
|
||||
// The subject list is complete and the chunk size is fixed, so the number
|
||||
// of batch queries is known before the first one flies. onProgress fires
|
||||
// after every answered batch with { done, total }.
|
||||
const total = groups.length * Math.ceil(subjects.length / SUBJECT_CHUNK_SIZE);
|
||||
let done = 0;
|
||||
for (const group of groups) {
|
||||
for (let i = 0; i < subjects.length; i += SUBJECT_CHUNK_SIZE) {
|
||||
const chunk = subjects.slice(i, i + SUBJECT_CHUNK_SIZE);
|
||||
const result = await sessionQuery(groupQuery(group, chunk));
|
||||
collect(result, group, values);
|
||||
done += 1;
|
||||
onProgress?.({ done, total });
|
||||
}
|
||||
}
|
||||
return values;
|
||||
|
||||
@@ -6,18 +6,14 @@ import { registerFile, registerReport } from './report.js';
|
||||
import { startShape } from './profile.js';
|
||||
|
||||
// Ties the steps together: seed → columns → assemble → CSV → register.
|
||||
// Returns { reportUri, fileUri, rowCount }. onProgress fires on every
|
||||
// answered query with what the phase knows — { subjects } per seed page,
|
||||
// { done, total } per column batch — so a waiter can tell a live run from
|
||||
// a stalled one.
|
||||
// Returns { reportUri, fileUri, rowCount }.
|
||||
|
||||
export async function run(query, update, parsed, profile, title, extra = {}, onProgress = null) {
|
||||
export async function run(query, update, parsed, profile, title, extra = {}) {
|
||||
const shape = startShape(profile, parsed);
|
||||
const subjects = await seed(query, parsed, shape,
|
||||
onProgress ? (n) => onProgress({ subjects: n }) : null);
|
||||
const subjects = await seed(query, parsed, shape);
|
||||
|
||||
const values = new Map();
|
||||
await fetchColumns(query, subjects, parsed, values, null, onProgress);
|
||||
await fetchColumns(query, subjects, parsed, values);
|
||||
|
||||
const rows = assemble(subjects, values, parsed);
|
||||
const fileName = extra.fileName || `${slug(title)}.csv`;
|
||||
|
||||
@@ -1,12 +1,10 @@
|
||||
import { sparqlEscapeUri, sparqlEscapeString, sparqlEscapeDateTime, sparqlEscapeDate } from '../db.js';
|
||||
|
||||
// Subject selection. One paged query per page; stops at ROW_LIMIT.
|
||||
// onPage(n) fires after every answered page with the subject count so far,
|
||||
// so a caller waiting on this run can tell a live one from a stalled one.
|
||||
const ROW_LIMIT = Number(process.env.ROW_LIMIT || 200000);
|
||||
const PAGE_SIZE = Number(process.env.SEED_PAGE_SIZE || 5000);
|
||||
|
||||
export async function seed(sessionQuery, spec, shape = null, onPage = null) {
|
||||
export async function seed(sessionQuery, spec, shape = null) {
|
||||
const subjects = [];
|
||||
let offset = 0;
|
||||
for (;;) {
|
||||
@@ -14,7 +12,6 @@ export async function seed(sessionQuery, spec, shape = null, onPage = null) {
|
||||
const result = await sessionQuery(query);
|
||||
const rows = result.results.bindings.map(b => b.s.value);
|
||||
subjects.push(...rows);
|
||||
onPage?.(subjects.length);
|
||||
if (subjects.length > ROW_LIMIT) {
|
||||
throw new Error(`the question matches more than ${ROW_LIMIT} subjects. Add a filter, for example a date range or a bestuurseenheid.`);
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user