Skip to content

feat(sparql-anything): convert several chunks at once #795

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
ddeboer merged 2 commits into from
Aug 31, 2026
Merged

feat(sparql-anything): convert several chunks at once #795

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
5086c30
feat(sparql-anything): convert several chunks at once
ddeboer Aug 31, 2026
f312c65
fix(sparql-anything): keep the abort path from outliving or hiding th…
ddeboer Aug 31, 2026
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
28 changes: 21 additions & 7 deletions docs/reference/sparql-anything.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -41,13 +41,14 @@ 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) |
| `taskRunner` | `TaskRunner<Task>` | 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) |
| `taskRunner` | `TaskRunner<Task>` | Runs the SPARQL Anything process for each chunk |

### Jobs

Expand Down Expand Up @@ -85,6 +86,19 @@ heap: '4g', // -Xmx4g

Size it with the chunk size. A chunk that outgrows the heap fails loudly – the JVM's `OutOfMemoryError` arrives in the output of a non-zero exit, which aborts the conversion – where an uncapped JVM instead grows until the OOM killer takes the whole container.

### Converting several chunks at once

`concurrency` is how many chunks are converted at the same time. Each one is a JVM of its own, so it multiplies against `heap`: a run needs `concurrency × heap`, on the machine the **task runner** uses – which is not this process's machine when the runner is Docker or remote.

That is why the default is `1` rather than something derived from the CPU count or a memory limit: the converter cannot see the machine its processes run on, so the number is the caller's to choose. `map.sh` sizes its pool from `nproc` capped by the cgroup limit, budgeting ~3 GB a worker; a caller who knows their deployment can do the same arithmetic and pass the result.

Chunks of every job are converted through one pool, in the order the jobs and their chunks were given – a long job and a short one pack together rather than draining in phases. The output is concatenated in that same order, however the processes happened to finish.

The first failure aborts the run: no further chunk is started, and the processes still going are stopped rather than left writing into a directory the converter is about to delete. A process that cannot be stopped – one that has just exited, say – does not change what is reported: the conversion failure is the one worth reading.

> [!WARNING]
> A `DockerTaskRunner` configured with a `containerName` cannot be used with `concurrency` above one. It force-removes any container of that name before starting a task, so each chunk would destroy the container of the chunk before it. Leave `containerName` unset for a converter that runs chunks in parallel.

`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.

## How a conversion runs
Expand Down
188 changes: 160 additions & 28 deletions packages/sparql-anything/src/sparql-anything-converter.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -84,6 +84,14 @@ export interface SparqlAnythingConverterOptions<Task> {
* are the converter's own, and are rejected here.
*/
cliArgs?: string[];
/**
* How many chunks to convert at once. Each one is a JVM of its own, so this
* multiplies against {@link heap}: the memory a run needs is `concurrency ×
* heap`, and the machine that has to hold it is the task runner's, not this
* process's. Left at one, chunks are converted one after another.
* @default 1
*/
concurrency?: number;
/** Runs the SPARQL Anything process for each chunk. */
taskRunner: TaskRunner<Task>;
}
Expand All @@ -98,6 +106,7 @@ export class SparqlAnythingConverter<Task> {
private readonly workDir: string;
private readonly heap: string;
private readonly cliArgs: string[];
private readonly concurrency: number;
private readonly taskRunner: TaskRunner<Task>;

constructor(options: SparqlAnythingConverterOptions<Task>) {
Expand All @@ -120,6 +129,13 @@ export class SparqlAnythingConverter<Task> {
);
}
this.cliArgs = options.cliArgs ?? [];
const concurrency = options.concurrency ?? 1;
if (!Number.isInteger(concurrency) || concurrency < 1) {
throw new Error(
`‘${concurrency}’ is not a number of chunks to convert at once; give a whole number of one or more`,
);
}
this.concurrency = concurrency;
this.taskRunner = options.taskRunner;
}

Expand All @@ -146,38 +162,114 @@ export class SparqlAnythingConverter<Task> {
const runDir = await mkdtemp(join(this.workDir, 'sparql-anything-'));
const runDirName = basename(runDir);
try {
const outputs: string[] = [];
let index = 0;
for (const { job, query } of planned) {
// Interpolated per chunk, at the point of use: holding a copy of the
// query for every chunk up front would scale with the input.
for (const chunk of job.chunks ?? [undefined]) {
const queryPath = join(runDirName, `query-${index}.rq`);
await writeFile(
join(this.workDir, queryPath),
chunk === undefined
? query
: // A replacer function, so `$&` and friends in a chunk path are
// the characters they look like rather than replacement patterns.
query.replaceAll(SOURCE_PLACEHOLDER, () => chunk),
);
const processOutput = join(runDirName, `output-${index}.nt`);
const task = await this.taskRunner.run(
this.command(queryPath, processOutput, job),
);
// wait() rejects on a non-zero exit, aborting convert() before the
// crashed process's missing output can be silently concatenated.
await this.taskRunner.wait(task);
const processOutputPath = join(this.workDir, processOutput);
await assertNonEmpty(processOutputPath, job, chunk);
outputs.push(processOutputPath);
index++;
const count = await this.runAll(planned, runDirName);
// By index, not by completion: the order the jobs and their chunks were
// given is the order of the triples, however the processes finished.
await concatenate(
Array.from({ length: count }, (_, index) =>
join(this.workDir, runDirName, `output-${index}.nt`),
),
outputPath,
);
} finally {
await rm(runDir, { recursive: true, force: true });
}
}

/**
* Runs every chunk, at most {@link concurrency} at a time, and returns how
* many processes that was.
*
* The first failure aborts the run: no further chunk is started, and the
* processes still going are stopped rather than left writing into a
* directory this is about to delete.
*/
private async runAll(
planned: PlannedJob[],
runDirName: string,
): Promise<number> {
const pending = processesOf(planned);
const state: RunState<Task> = { 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
// for the others, whatever the reason it stopped.
const convertChunks = async (): Promise<void> => {
while (state.failure === undefined) {
const next = pending.next();
if (next.done === true) {
return;
}
try {
await this.convertChunk(next.value, runDirName, state);
} catch (error) {
state.failure ??= error;
await this.stopInFlight(state);
return;
}
}
};

await Promise.all(
Array.from({ length: this.concurrency }, () => convertChunks()),
);
if (state.failure !== undefined) {
throw state.failure;
}
return countOf(planned);
}

/** Converts one chunk, writing `output-<index>.nt` in the run directory. */
private async convertChunk(
{ index, job, chunk, query }: PlannedProcess,
runDirName: string,
state: RunState<Task>,
): Promise<void> {
const queryPath = join(runDirName, `query-${index}.rq`);
await writeFile(
join(this.workDir, queryPath),
chunk === undefined
? query
: // A replacer function, so `$&` and friends in a chunk path are the
// characters they look like rather than replacement patterns.
query.replaceAll(SOURCE_PLACEHOLDER, () => chunk),
);
const output = join(runDirName, `output-${index}.nt`);
const task = await this.taskRunner.run(
this.command(queryPath, output, job),
);
state.inFlight.add(task);
try {
if (state.failure !== undefined) {
// Started in the window between another chunk failing and this worker
// seeing it, so it is not in the set that failure stopped.
await this.stopQuietly(task);
return;
}
await concatenate(outputs, outputPath);
// wait() rejects on a non-zero exit, aborting convert() before the
// crashed chunk's missing output can be silently concatenated.
await this.taskRunner.wait(task);
} finally {
await rm(runDir, { recursive: true, force: true });
state.inFlight.delete(task);
}
await assertNonEmpty(join(this.workDir, output), job, chunk);
}

/** Stops every process still running, so none outlives the run. */
private async stopInFlight(state: RunState<Task>): Promise<void> {
await Promise.all(
[...state.inFlight].map((task) => this.stopQuietly(task)),
);
}

/**
* Stops a task, ignoring a failure to stop it. Stopping is best effort by
* nature – a process that has just exited cannot be stopped, and reporting
* that would replace the failure that is actually worth reporting, and leave
* the other workers unawaited while the run directory is deleted.
*/
private async stopQuietly(task: Task): Promise<void> {
await this.taskRunner.stop(task).catch(() => undefined);
}

/** The SPARQL Anything invocation for one job. */
Expand All @@ -203,6 +295,46 @@ export class SparqlAnythingConverter<Task> {
}
}

/** One SPARQL Anything process: a job's query, to run over one of its chunks. */
interface PlannedProcess {
/** Position in the run, which orders the outputs and names their files. */
index: number;
job: ConversionJob;
chunk?: string;
/** The job's query as written, with `{SOURCE}` still in it. */
query: string;
}

/**
* The processes the planned jobs call for, in order: one per chunk, and one
* for a job that has none. Lazy, so the workers pulling from it hold one
* process each rather than the whole run.
*/
function* processesOf(planned: PlannedJob[]): Generator<PlannedProcess> {
let index = 0;
for (const { job, query } of planned) {
for (const chunk of job.chunks ?? [undefined]) {
yield { index: index++, job, chunk, query };
}
}
}

/** How many processes the planned jobs call for. */
function countOf(planned: PlannedJob[]): number {
return planned.reduce(
(total, { job }) => total + (job.chunks?.length ?? 1),
0,
);
}

/** What the workers of one run share. */
interface RunState<Task> {
/** Tasks that have been started and not yet finished. */
inFlight: Set<Task>;
/** The first failure, which aborts the run. */
failure?: unknown;
}

/** A job whose query has been read, and checked against its chunks. */
interface PlannedJob {
job: ConversionJob;
Expand Down
Loading