diff --git a/docs/reference/sparql-anything.md b/docs/reference/sparql-anything.md index 262e8be4..ef915e4e 100644 --- a/docs/reference/sparql-anything.md +++ b/docs/reference/sparql-anything.md @@ -41,14 +41,15 @@ await converter.convert( ### Options -| Option | Type | Description | -| ------------- | ------------------ | ----------------------------------------------------------------------------------------------------------------------------- | -| `jarPath` | `string` | Path to the SPARQL Anything CLI jar, as the task runner sees it | -| `workDir` | `string` | The task runner's working directory; see [Where files are written](#where-files-are-written) | -| `heap` | `string` | Maximum JVM heap per chunk process, as `-Xmx` takes it (default `'2g'`); see [Memory](#memory) | -| `cliArgs` | `string[]` | Further arguments for the SPARQL Anything CLI; see [Memory](#memory) | -| `concurrency` | `number` | How many chunks to convert at once (default `1`); see [Converting several chunks at once](#converting-several-chunks-at-once) | -| `taskRunner` | `TaskRunner` | Runs the SPARQL Anything process for each chunk | +| Option | Type | Description | +| ------------------ | -------------------- | ----------------------------------------------------------------------------------------------------------------------------- | +| `jarPath` | `string` | Path to the SPARQL Anything CLI jar, as the task runner sees it | +| `workDir` | `string` | The task runner's working directory; see [Where files are written](#where-files-are-written) | +| `heap` | `string` | Maximum JVM heap per chunk process, as `-Xmx` takes it (default `'2g'`); see [Memory](#memory) | +| `cliArgs` | `string[]` | Further arguments for the SPARQL Anything CLI; see [Memory](#memory) | +| `concurrency` | `number` | How many chunks to convert at once (default `1`); see [Converting several chunks at once](#converting-several-chunks-at-once) | +| `onChunkConverted` | `(progress) => void` | Called as each chunk finishes; see [Following a conversion](#following-a-conversion) | +| `taskRunner` | `TaskRunner` | Runs the SPARQL Anything process for each chunk | ### Jobs @@ -101,6 +102,19 @@ The first failure aborts the run: no further chunk is started, and the processes `cliArgs` is the escape hatch for CLI flags the converter does not model itself, appended to the arguments it sets. It cannot repeat those: `-q`, `-f`, `-o` and `-l` are rejected, in their long and `--flag=value` forms too, because the converter reads back the `--output` it named, in the `--format` it asked for – overriding either leaves it reporting an empty conversion, or concatenating fragments that are not N-Triples. SPARQL Anything documents repetition only for `-v` and `-c`, so a repeated flag has no defined winner to rely on. +### Following a conversion + +A conversion says nothing for as long as it takes – a quarter of an hour, over eighteen chunks, for the GeoNames dumps. `onChunkConverted` is called as each one finishes: + +```typescript +onChunkConverted: ({ index, total, chunk }) => + console.log(`Converted ${index}/${total}${chunk === undefined ? '' : `: ${chunk}`}`), +``` + +It is called once per chunk, in the order they finish rather than the order they were given, and not at all for a chunk that failed – a failure arrives as the rejection instead. A callback that throws aborts the run, like any other failure. + +It is wiring rather than configuration: a runtime that already has somewhere to report progress passes a function that forwards to it, which is why this is a plain callback rather than a reporter interface of its own. + ## Chunking `convert()` takes chunks it does not create, because a caller often has work to do first – filtering rows out, say. `chunk()` produces them: diff --git a/packages/sparql-anything/src/index.ts b/packages/sparql-anything/src/index.ts index f7e27c7f..220b0d4e 100644 --- a/packages/sparql-anything/src/index.ts +++ b/packages/sparql-anything/src/index.ts @@ -1,6 +1,7 @@ export { chunk, type ChunkOptions } from './chunk.js'; export { SparqlAnythingConverter, + type ChunkProgress, type ConversionJob, type SparqlAnythingConverterOptions, } from './sparql-anything-converter.js'; diff --git a/packages/sparql-anything/src/sparql-anything-converter.ts b/packages/sparql-anything/src/sparql-anything-converter.ts index 5c1c8a29..649fac10 100644 --- a/packages/sparql-anything/src/sparql-anything-converter.ts +++ b/packages/sparql-anything/src/sparql-anything-converter.ts @@ -58,6 +58,18 @@ export interface ConversionJob { load?: string; } +/** What a chunk's conversion reports when it is done. */ +export interface ChunkProgress { + /** Position of this process in the run, counting from one. */ + index: number; + /** How many processes the run holds in all. */ + total: number; + /** The chunk converted, for a job that has chunks. */ + chunk?: string; + /** The query the job ran, which is what tells two jobs apart. */ + queryFile: string; +} + /** Configuration for a {@link SparqlAnythingConverter}. */ export interface SparqlAnythingConverterOptions { /** Path to the SPARQL Anything CLI jar, as the task runner sees it. */ @@ -94,6 +106,14 @@ export interface SparqlAnythingConverterOptions { concurrency?: number; /** Runs the SPARQL Anything process for each chunk. */ taskRunner: TaskRunner; + /** + * Called as each chunk finishes, for a conversion that would otherwise say + * nothing for as long as it takes – the GeoNames run is a quarter of an hour + * over eighteen chunks. Called once per chunk, in the order they finish + * rather than the order they were given, and not at all for a chunk that + * failed. A callback that throws aborts the run, like any other failure. + */ + onChunkConverted?: (progress: ChunkProgress) => void; } /** @@ -108,6 +128,7 @@ export class SparqlAnythingConverter { private readonly cliArgs: string[]; private readonly concurrency: number; private readonly taskRunner: TaskRunner; + private readonly onChunkConverted?: (progress: ChunkProgress) => void; constructor(options: SparqlAnythingConverterOptions) { this.jarPath = options.jarPath; @@ -137,6 +158,7 @@ export class SparqlAnythingConverter { } this.concurrency = concurrency; this.taskRunner = options.taskRunner; + this.onChunkConverted = options.onChunkConverted; } /** @@ -189,7 +211,10 @@ export class SparqlAnythingConverter { runDirName: string, ): Promise { const pending = processesOf(planned); - const state: RunState = { inFlight: new Set() }; + const state: RunState = { + total: countOf(planned), + inFlight: new Set(), + }; // Pulled one at a time rather than with `for...of`: leaving a for-of early // closes the iterator, so the first worker to give up would end the queue @@ -216,7 +241,7 @@ export class SparqlAnythingConverter { if (state.failure !== undefined) { throw state.failure; } - return countOf(planned); + return state.total; } /** Converts one chunk, writing `output-.nt` in the run directory. */ @@ -253,6 +278,12 @@ export class SparqlAnythingConverter { state.inFlight.delete(task); } await assertNonEmpty(join(this.workDir, output), job, chunk); + this.onChunkConverted?.({ + index: index + 1, + total: state.total, + chunk, + queryFile: job.queryFile, + }); } /** Stops every process still running, so none outlives the run. */ @@ -329,6 +360,8 @@ function countOf(planned: PlannedJob[]): number { /** What the workers of one run share. */ interface RunState { + /** How many processes the run holds, for what reports progress. */ + total: number; /** Tasks that have been started and not yet finished. */ inFlight: Set; /** The first failure, which aborts the run. */ diff --git a/packages/sparql-anything/test/sparql-anything-converter.test.ts b/packages/sparql-anything/test/sparql-anything-converter.test.ts index 6eb2feb8..68c09c2d 100644 --- a/packages/sparql-anything/test/sparql-anything-converter.test.ts +++ b/packages/sparql-anything/test/sparql-anything-converter.test.ts @@ -1,4 +1,8 @@ -import { ConversionJob, SparqlAnythingConverter } from '../src/index.js'; +import { + ChunkProgress, + ConversionJob, + SparqlAnythingConverter, +} from '../src/index.js'; import { TaskRunner } from '@lde/task-runner'; import { describe, it, expect, beforeEach, afterEach } from 'vitest'; import { @@ -620,6 +624,55 @@ describe('SparqlAnythingConverter', () => { ).toThrow('is not a number of chunks to convert at once'); }); + it('reports each chunk as it finishes', async () => { + const taskRunner = new FakeTaskRunner(workDir); + const chunks = await writeChunks(2); + const ontologyQuery = join(workDir, 'ontology.rq'); + await writeFile(ontologyQuery, 'CONSTRUCT { ?s ?p ?o } WHERE { ?s ?p ?o }'); + const reported: ChunkProgress[] = []; + + await new SparqlAnythingConverter({ + jarPath: '/bin/sparql-anything.jar', + workDir, + taskRunner, + onChunkConverted: (progress) => reported.push(progress), + }).convert( + [ + { queryFile, chunks }, + { queryFile: ontologyQuery, load: '/data/ontology.rdf' }, + ], + join(workDir, 'output.nt'), + ); + + expect(reported).toEqual([ + { index: 1, total: 3, chunk: chunks[0], queryFile }, + { index: 2, total: 3, chunk: chunks[1], queryFile }, + // A job without chunks reports too, with none to name. + { index: 3, total: 3, chunk: undefined, queryFile: ontologyQuery }, + ]); + }); + + it('does not report a chunk that failed', async () => { + const chunks = await writeChunks(2); + const taskRunner = new FakeTaskRunner(workDir, { + failOutputContaining: 'output-1.nt', + }); + const reported: ChunkProgress[] = []; + + await expect( + new SparqlAnythingConverter({ + jarPath: '/bin/sparql-anything.jar', + workDir, + taskRunner, + onChunkConverted: (progress) => reported.push(progress), + }).convert([{ queryFile, chunks }], join(workDir, 'output.nt')), + ).rejects.toThrow('Process failed'); + + expect(reported).toEqual([ + { index: 1, total: 2, chunk: chunks[0], queryFile }, + ]); + }); + it('refuses an empty job list rather than writing an empty output', async () => { const taskRunner = new FakeTaskRunner(workDir); const outputPath = join(workDir, 'output.nt');