fix(sparql-anything): accept an empty output from a readable, non-empty chunk

docs/reference/sparql-anything.md

@@ -69,9 +69,9 @@ A query and its chunks have to agree: a query naming `{SOURCE}` without chunks,
 
 `workDir` is the task runner's working directory – `cwd` for a `NativeTaskRunner`, `mountDir` for a `DockerTaskRunner`. The converter writes its generated query files and per-process outputs into a fresh subdirectory there and removes it when the conversion ends, then refers to them by a path relative to `workDir`, so the identical command works on the host and inside a container.
 
-Per-run directories matter for more than tidiness: an output left over from an earlier run would satisfy the non-empty check below with stale triples.
+Per-run directories matter for more than tidiness: an output left over from an earlier run would satisfy the output check below with stale triples.
 
-`jarPath`, and each job's `load` and `chunks` paths, are passed through as given, because only the caller knows how the runner sees them – in a container the jar usually lives in the image, while the chunks have to be under the mount.
+`jarPath`, and each job's `load` and `chunks` paths, are passed through as given, because only the caller knows how the runner sees them – in a container the jar usually lives in the image, while the chunks have to be under the mount. A **relative** `load` or chunk path is relative to `workDir` for the runner and for the converter alike, so the converter checks it before starting a process: one that is missing or empty fails then, naming the file, before a JVM is spent on it. An **absolute** path is the runner's, and one the converter cannot find on its own side proves nothing either way – the process runs, and its output is held to the stricter check below.
 
 ### Loading existing RDF
 
@@ -155,7 +155,7 @@ For each chunk – or once, for a job that has none – the converter:
 1. Replaces the literal `{SOURCE}` in the job’s query with the chunk’s path and writes the result to a temporary `.rq` file. The query is read once per job, and interpolated per chunk.
 2. Runs `java -Xmx<heap> -jar <jar> -q <query> [--load <load>] --format NT --output <chunk>.nt [cliArgs]`, with every path quoted, so a space or a shell metacharacter in a filename can neither break the command nor inject into it.
 3. Waits for the process; a non-zero exit **aborts the whole conversion** so a crashed chunk can never be silently dropped from the output.
-4. Checks that the output is not empty. SPARQL Anything exits successfully when it cannot read or parse an input – it logs the problem and writes nothing – so an empty or missing output **aborts the conversion** too.
+4. Checks the output. SPARQL Anything exits successfully when it cannot read or parse an input – it logs the problem, writes an empty output and stops – but a chunk whose rows the query filters out entirely writes an empty output too. The converter tells the two apart by what it saw before starting: when the chunk and the job's `load`, if any, were both found on its own side and non-empty, an empty output is **accepted as no triples** and contributes nothing to the concatenation. Otherwise – an input the converter could not see, or a job without chunks, whose input the query names – an empty output **aborts the conversion**, and a missing output always does.
 
 Converting an empty list of jobs is an error rather than an empty output: a step that produced none has already failed.
 

packages/sparql-anything/src/sparql-anything-converter.ts

@@ -2,7 +2,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 { basename, join } from 'node:path';
+import { basename, isAbsolute, join } from 'node:path';
 
 /** Placeholder in the query file that is replaced with each chunk's path. */
 const SOURCE_PLACEHOLDER = '{SOURCE}';
@@ -167,7 +167,8 @@ export class SparqlAnythingConverter<Task> {
    *
    * `queryFile` is read here, so it must be readable by this process; the chunk
    * and `load` paths are passed to SPARQL Anything as given, so those must be
-   * readable by the task runner.
+   * readable by the task runner. A relative one is resolved against `workDir`
+   * on both sides, so it is checked here too, before a process is spent on it.
    *
    * Jobs of different shapes belong in one call: they are run by one converter,
    * so a long job and a short one pack together instead of draining in phases.
@@ -260,6 +261,16 @@ export class SparqlAnythingConverter<Task> {
           query.replaceAll(SOURCE_PLACEHOLDER, () => chunk),
     );
     const output = join(runDirName, `output-${index}.nt`);
+    // Before the process rather than after it: a missing or empty input fails
+    // now, naming the file, and only what was seen to be non-empty can excuse
+    // an empty output. A job without chunks names its input in the query,
+    // which the converter cannot see, so its output has to be non-empty.
+    const inputsVerified =
+      chunk !== undefined &&
+      (await verifyInputs(
+        [chunk, ...(job.load === undefined ? [] : [job.load])],
+        this.workDir,
+      ));
     const task = await this.taskRunner.run(
       this.command(queryPath, output, job),
     );
@@ -277,7 +288,12 @@ export class SparqlAnythingConverter<Task> {
     } finally {
       state.inFlight.delete(task);
     }
-    await assertNonEmpty(join(this.workDir, output), job, chunk);
+    await assertNonEmpty(
+      join(this.workDir, output),
+      job,
+      chunk,
+      inputsVerified,
+    );
     this.onChunkConverted?.({
       index: index + 1,
       total: state.total,
@@ -417,32 +433,74 @@ async function plan(jobs: ConversionJob[]): Promise<PlannedJob[]> {
 }
 
 /**
- * Throws unless the job's output holds at least one byte. SPARQL Anything
- * exits 0 when it cannot read or parse an input: it logs the problem, writes an
- * empty output and stops. Without this guard a run stays green while its output
- * silently misses every triple of the chunk.
+ * Whether every one of `paths` – a process's chunk, and its job's `load` – was
+ * seen from here to hold at least one byte, which is what tells an empty
+ * output that means “no triples” from one that means an unreadable input.
+ *
+ * A relative path is relative to `workDir` for the task runner as much as for
+ * this process, so one that is missing or empty there is an upstream failure,
+ * reported now rather than after a JVM has run over it. An absolute path is
+ * the runner's – under its mount, say – and one that does not exist from here
+ * proves nothing either way; such a process runs, and its output then has to
+ * be non-empty as before.
+ */
+async function verifyInputs(
+  paths: string[],
+  workDir: string,
+): Promise<boolean> {
+  let verified = true;
+  for (const path of paths) {
+    const size = await sizeOf(isAbsolute(path) ? path : join(workDir, path));
+    if (size === undefined) {
+      if (!isAbsolute(path)) {
+        throw new Error(
+          `Input ‘${path}’ does not exist under ‘${workDir}’; a step that should have produced it has failed upstream`,
+        );
+      }
+      verified = false;
+    } else if (size === 0) {
+      throw new Error(
+        `Input ‘${path}’ is empty; a step that produced it has failed upstream, and converting nothing would hide that`,
+      );
+    }
+  }
+  return verified;
+}
+
+/**
+ * Throws unless the job's output holds at least one byte, or is empty for a
+ * chunk whose inputs were all verified, which is a chunk the query filtered out
+ * entirely. SPARQL Anything exits 0 when it cannot read or parse an input: it
+ * logs the problem, writes an empty output and stops. Without this guard a run
+ * stays green while its output silently misses every triple of the chunk.
  */
 async function assertNonEmpty(
   outputPath: string,
   job: ConversionJob,
-  chunk?: string,
+  chunk: string | undefined,
+  inputsVerified: boolean,
 ): Promise<void> {
-  const size = await stat(outputPath).then(
+  const size = await sizeOf(outputPath);
+  if (size === undefined || (size === 0 && !inputsVerified)) {
+    throw new Error(
+      `SPARQL Anything produced no output for ‘${job.queryFile}’${chunk === undefined ? '' : ` over ‘${chunk}’`}; it exits successfully when it cannot read or parse an input`,
+    );
+  }
+}
+
+/** The size of the file at `path` in bytes, or undefined when there is none. */
+async function sizeOf(path: string): Promise<number | undefined> {
+  return stat(path).then(
     (stats) => stats.size,
     (error: NodeJS.ErrnoException) => {
       // Anything but a missing file is a problem of its own, and reporting it
       // as an empty conversion would send the reader after the wrong cause.
       if (error.code !== 'ENOENT') {
         throw error;
       }
-      return 0;
+      return undefined;
     },
   );
-  if (size === 0) {
-    throw new Error(
-      `SPARQL Anything produced no output for ‘${job.queryFile}’${chunk === undefined ? '' : ` over ‘${chunk}’`}; it exits successfully when it cannot read or parse an input`,
-    );
-  }
 }
 
 /**

packages/sparql-anything/test/sparql-anything-converter.test.ts

@@ -471,6 +471,23 @@ describe('SparqlAnythingConverter', () => {
     ).rejects.toThrow(/produced no output for ‘.*ontology\.rq’;/);
   });
 
+  it('aborts when a chunkless job without --load produces no output', async () => {
+    // The query names its own input, which the converter cannot see, so an
+    // empty output cannot be told from an unreadable input.
+    const taskRunner = new FakeTaskRunner(workDir, {
+      emptyOutputContaining: 'output-0.nt',
+    });
+    const ontologyQuery = join(workDir, 'ontology.rq');
+    await writeFile(ontologyQuery, 'CONSTRUCT { ?s ?p ?o } WHERE { ?s ?p ?o }');
+
+    await expect(
+      converterFor(taskRunner).convert(
+        [{ queryFile: ontologyQuery }],
+        join(workDir, 'output.nt'),
+      ),
+    ).rejects.toThrow('produced no output');
+  });
+
   it('refuses a job whose query names {SOURCE} but has no chunk', async () => {
     const taskRunner = new FakeTaskRunner(workDir);
 
@@ -685,8 +702,40 @@ describe('SparqlAnythingConverter', () => {
     await expect(readFile(outputPath, 'utf-8')).rejects.toThrow();
   });
 
-  it('aborts when a chunk produces an empty output', async () => {
+  it('accepts an empty output from a chunk it saw to be non-empty, as one the query filtered out', async () => {
     const chunks = await writeChunks(3);
+    const taskRunner = new FakeTaskRunner(workDir, {
+      emptyOutputContaining: 'output-1.nt',
+    });
+    const outputPath = join(workDir, 'output.nt');
+    const converted: ChunkProgress[] = [];
+
+    await new SparqlAnythingConverter({
+      jarPath: '/bin/sparql-anything.jar',
+      workDir,
+      taskRunner,
+      onChunkConverted: (progress) => converted.push(progress),
+    }).convert(jobsFor(chunks), outputPath);
+
+    // The filtered chunk contributes nothing to the concatenation; the others
+    // are unaffected, and it is reported as converted like any other.
+    expect(taskRunner.commands).toHaveLength(3);
+    const output = await readFile(outputPath, 'utf-8');
+    expect(output).toMatch(/output-0\.nt/);
+    expect(output).not.toMatch(/output-1\.nt/);
+    expect(output).toMatch(/output-2\.nt/);
+    expect(converted.map(({ chunk }) => chunk)).toEqual(chunks);
+  });
+
+  it('aborts when a chunk it could not see produces an empty output', async () => {
+    // Absolute paths the task runner sees but this process does not – under a
+    // container's mount, say – so nothing tells a filtered chunk from one
+    // SPARQL Anything could not read, and the guard stays as it was.
+    const chunks = [
+      '/mount/chunk-0.csv',
+      '/mount/chunk-1.csv',
+      '/mount/chunk-2.csv',
+    ];
     // SPARQL Anything exits 0 but writes nothing when it cannot read an input.
     const taskRunner = new FakeTaskRunner(workDir, {
       emptyOutputContaining: 'output-1.nt',
@@ -702,6 +751,51 @@ describe('SparqlAnythingConverter', () => {
     await expect(readFile(outputPath, 'utf-8')).rejects.toThrow();
   });
 
+  it('refuses a relative chunk path that does not exist under workDir before starting anything', async () => {
+    const taskRunner = new FakeTaskRunner(workDir);
+
+    await expect(
+      converterFor(taskRunner).convert(
+        jobsFor(['chunks/missing.csv']),
+        join(workDir, 'output.nt'),
+      ),
+    ).rejects.toThrow(
+      `Input ‘chunks/missing.csv’ does not exist under ‘${workDir}’`,
+    );
+
+    expect(taskRunner.commands).toHaveLength(0);
+  });
+
+  it('refuses an empty chunk before starting anything', async () => {
+    const taskRunner = new FakeTaskRunner(workDir);
+    const [chunk] = await writeChunks(1);
+    await writeFile(chunk, '');
+
+    await expect(
+      converterFor(taskRunner).convert(
+        jobsFor([chunk]),
+        join(workDir, 'output.nt'),
+      ),
+    ).rejects.toThrow(`Input ‘${chunk}’ is empty`);
+
+    expect(taskRunner.commands).toHaveLength(0);
+  });
+
+  it('refuses an empty --load path before starting anything', async () => {
+    const taskRunner = new FakeTaskRunner(workDir);
+    const [chunk] = await writeChunks(1);
+    await writeFile(join(workDir, 'reference.ttl'), '');
+
+    await expect(
+      converterFor(taskRunner).convert(
+        jobsFor([chunk], 'reference.ttl'),
+        join(workDir, 'output.nt'),
+      ),
+    ).rejects.toThrow('Input ‘reference.ttl’ is empty');
+
+    expect(taskRunner.commands).toHaveLength(0);
+  });
+
   it('aborts when a chunk produces no output file', async () => {
     const chunks = await writeChunks(2);
     const taskRunner = new FakeTaskRunner(workDir, {