diff --git a/docs/reference/sparql-anything.md b/docs/reference/sparql-anything.md index ef915e4e..240aabe2 100644 --- a/docs/reference/sparql-anything.md +++ b/docs/reference/sparql-anything.md @@ -159,4 +159,4 @@ For each chunk – or once, for a job that has none – the converter: Converting an empty list of jobs is an error rather than an empty output: a step that produced none has already failed. -Finally, the `.nt` files are concatenated, in the order the jobs and their chunks were given, into the output path. The concatenation streams, so multi-gigabyte outputs do not have to fit in memory. N-Triples has no prefixes or document structure, so concatenating per-chunk files always yields a single valid document. +Finally, the `.nt` files are concatenated, in the order the jobs and their chunks were given, into the output path. The concatenation streams, so multi-gigabyte outputs do not have to fit in memory, and it is byte for byte what `cat` would give: a newline is inserted only after a file that does not end in one, so two triples never share a line and no blank line is added between files. N-Triples has no prefixes or document structure, so concatenating per-chunk files always yields a single valid document. diff --git a/packages/sparql-anything/src/sparql-anything-converter.ts b/packages/sparql-anything/src/sparql-anything-converter.ts index 649fac10..ed41ccf7 100644 --- a/packages/sparql-anything/src/sparql-anything-converter.ts +++ b/packages/sparql-anything/src/sparql-anything-converter.ts @@ -1,7 +1,7 @@ import { shellQuote, TaskRunner } from '@lde/task-runner'; import { createReadStream, createWriteStream } from 'node:fs'; import { mkdtemp, readFile, rm, stat, writeFile } from 'node:fs/promises'; -import { finished, pipeline } from 'node:stream/promises'; +import { pipeline } from 'node:stream/promises'; import { basename, join } from 'node:path'; /** Placeholder in the query file that is replaced with each chunk's path. */ @@ -445,27 +445,37 @@ async function assertNonEmpty( } } +/** The byte that ends an N-Triples line. */ +const NEWLINE = '\n'.charCodeAt(0); + /** * Concatenates `inputPaths` into `outputPath`, streaming so multi-GB outputs do * not have to fit in memory. N-Triples has no prefixes or document structure, so * concatenating per-chunk files yields a single valid document. + * + * One pipeline over every input rather than one per input into a shared + * destination: each pipeline() leaves its listeners on the destination, and + * past ten of them Node warns of a leak. */ async function concatenate( inputPaths: string[], outputPath: string, ): Promise { - const output = createWriteStream(outputPath); - for (const [index, inputPath] of inputPaths.entries()) { - // A newline between files, in case one does not end in one: N-Triples - // tolerates the blank line, but not two triples sharing a line. - if (index > 0) { - output.write('\n'); + await pipeline(async function* () { + let endsInNewline = true; + for (const inputPath of inputPaths) { + // A newline between files only when one does not end in one, so the + // result is byte for byte what `cat` gives: N-Triples tolerates a blank + // line, but not two triples sharing a line. + if (!endsInNewline) { + yield '\n'; + } + for await (const chunk of createReadStream( + inputPath, + ) as AsyncIterable) { + yield chunk; + endsInNewline = chunk.at(-1) === NEWLINE; + } } - await pipeline(createReadStream(inputPath), output, { end: false }); - } - output.end(); - // pipeline() with `end: false` resolves once the source ends, not once the - // destination is flushed and closed, so await that before reporting success. - // On an earlier rejection pipeline() has already destroyed the stream. - await finished(output); + }, createWriteStream(outputPath)); } diff --git a/packages/sparql-anything/test/sparql-anything-converter.test.ts b/packages/sparql-anything/test/sparql-anything-converter.test.ts index 68c09c2d..2ac7f267 100644 --- a/packages/sparql-anything/test/sparql-anything-converter.test.ts +++ b/packages/sparql-anything/test/sparql-anything-converter.test.ts @@ -30,6 +30,11 @@ interface FakeTaskRunnerOptions { * all, as SPARQL Anything leaves behind when it crashes before writing. */ missingOutputContaining?: string; + /** + * When set, chunks whose output path contains this get an `--output` that + * does not end in a newline, as a writer other than Jena’s might leave. + */ + unterminatedOutputContaining?: string; /** * When set, chunks whose output path contains this get an `--output` that * cannot be inspected – a symlink to itself, which fails `stat` with ELOOP. @@ -93,6 +98,7 @@ class FakeTaskRunner implements TaskRunner<{ command: string }> { const { emptyOutputContaining, missingOutputContaining, + unterminatedOutputContaining, unreadableOutputContaining, } = this.options; if (matches(outputFile, missingOutputContaining)) { @@ -105,7 +111,11 @@ class FakeTaskRunner implements TaskRunner<{ command: string }> { } await writeFile( path, - matches(outputFile, emptyOutputContaining) ? '' : `${outputFile}\n`, + matches(outputFile, emptyOutputContaining) + ? '' + : matches(outputFile, unterminatedOutputContaining) + ? outputFile + : `${outputFile}\n`, ); } @@ -386,10 +396,58 @@ describe('SparqlAnythingConverter', () => { // The FakeTaskRunner writes each chunk's output path as that file's // content, so the result reflects the order the chunks were processed. expect(await readFile(outputPath, 'utf-8')).toMatch( - /^sparql-anything-\S+\/output-0\.nt\n\nsparql-anything-\S+\/output-1\.nt\n\nsparql-anything-\S+\/output-2\.nt\n$/, + /^sparql-anything-\S+\/output-0\.nt\nsparql-anything-\S+\/output-1\.nt\nsparql-anything-\S+\/output-2\.nt\n$/, ); }); + it('concatenates outputs byte for byte, as cat would', async () => { + const taskRunner = new FakeTaskRunner(workDir); + const chunks = await writeChunks(3); + const outputPath = join(workDir, 'output.nt'); + + await converterFor(taskRunner).convert(jobsFor(chunks), outputPath); + + const outputs = taskRunner.commands.map( + (command) => `${tokenAfter(command, '--output')}\n`, + ); + expect(await readFile(outputPath, 'utf-8')).toBe(outputs.join('')); + }); + + it('separates an output that does not end in a newline from the next', async () => { + const taskRunner = new FakeTaskRunner(workDir, { + unterminatedOutputContaining: 'output-0.nt', + }); + const chunks = await writeChunks(2); + const outputPath = join(workDir, 'output.nt'); + + await converterFor(taskRunner).convert(jobsFor(chunks), outputPath); + + expect(await readFile(outputPath, 'utf-8')).toMatch( + /^sparql-anything-\S+\/output-0\.nt\nsparql-anything-\S+\/output-1\.nt\n$/, + ); + }); + + it('concatenates more than ten outputs without a listener leak warning', async () => { + const taskRunner = new FakeTaskRunner(workDir); + const chunks = await writeChunks(12); + const warnings: Error[] = []; + const onWarning = (warning: Error) => warnings.push(warning); + process.on('warning', onWarning); + + try { + await converterFor(taskRunner).convert( + jobsFor(chunks), + join(workDir, 'output.nt'), + ); + // Node emits process warnings on a later tick, so give them one. + await new Promise((resolve) => setImmediate(resolve)); + } finally { + process.off('warning', onWarning); + } + + expect(warnings).toEqual([]); + }); + it('leaves nothing behind in the working directory', async () => { const taskRunner = new FakeTaskRunner(workDir); const chunks = await writeChunks(2); @@ -540,7 +598,7 @@ describe('SparqlAnythingConverter', () => { }).convert([{ queryFile, chunks }], outputPath); expect(await readFile(outputPath, 'utf-8')).toMatch( - /^sparql-anything-\S+\/output-0\.nt\n\nsparql-anything-\S+\/output-1\.nt\n\nsparql-anything-\S+\/output-2\.nt\n$/, + /^sparql-anything-\S+\/output-0\.nt\nsparql-anything-\S+\/output-1\.nt\nsparql-anything-\S+\/output-2\.nt\n$/, ); });