From 9d5f6e32c72d5d104c8964ad65d305893730e29e Mon Sep 17 00:00:00 2001 From: David de Boer Date: Thu, 3 Sep 2026 19:33:43 +0200 Subject: [PATCH] feat(sparql-anything): report each chunk as it is converted MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A conversion says nothing for as long as it takes: a quarter of an hour over eighteen chunks for the GeoNames dumps, with the first sign of life being the output file or a rejection naming one chunk. onChunkConverted is called as each chunk finishes, with its position, how many there are in all, the chunk and the query that converted it. Once per chunk, in the order they finish rather than the order they were given, and not at all for a chunk that failed – that arrives as the rejection. A plain callback rather than a reporter interface. @lde/pipeline's ProgressReporter is shaped around datasets and distributions, so chunk progress has no event to be; taking one would couple this package to the pipeline, which its plugins depend on and this is not. A callback is also what a configurable runtime would inject: options are data a document can hold, collaborators are wiring the runtime supplies, and this is wiring. --- docs/reference/sparql-anything.md | 30 +++++++--- packages/sparql-anything/src/index.ts | 1 + .../src/sparql-anything-converter.ts | 37 ++++++++++++- .../test/sparql-anything-converter.test.ts | 55 ++++++++++++++++++- 4 files changed, 112 insertions(+), 11 deletions(-) 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');