Added generated quads to runtime output & more (#35)

* Updated Stage class's end event emit logic Added generated quads to runtime output Iterator's batchSize will now override the query limit when provided Updated README file Removed outdated comments
netwerk-digitaal-erfgoed · Jan 10, 2024 · 1963cc8 · 1963cc8
1 parent b2bc135
commit 1963cc8
Show file tree

Hide file tree

Showing 8 changed files with 100 additions and 38 deletions.
diff --git a/README.md b/README.md
@@ -6,8 +6,6 @@ LD Workbench is a Command Line Interface (CLI) application. LD Workbench is test
 
 This project is currently in a Proof-of-Concept phase. Feel free to watch our progress, but please do not use this project in a production setting.
 
-
-
 ## Approach
 
 A *pipeline* is the sequence of *stages*.
@@ -18,15 +16,59 @@ The *iterator* component is configured by a SPARQL Select query. This query bind
 
 Every binding for variable `$this` is used to parameterize a SPARQL Construct query; this is the *generator* component. Parameterization follows [SPARQL pre-binding](https://www.w3.org/TR/shacl/#pre-binding) according to the SHACL standard. Each SPARQL Construct query returns RDF triples that are part of the transformed result.
 
-
-
 ## Configuration
 
 An LD Workbench pipeline is defined with a YAML configuration file. The configuration is validated by a JSON Schema. The schema is part of this repository ([link](https://github.com/netwerk-digitaal-erfgoed/ld-workbench/blob/main/static/ld-workbench.schema.json)). The YAML and JSON Schema combination is tested to work in the VSCode editor.
 
 A pipeline must have a name, one or more stages, and optionally a description. Multiple pipelines can be configured as long as they have unique names. See the [example configuration file](https://github.com/netwerk-digitaal-erfgoed/ld-workbench/blob/main/static/example/config.yml) for a boilerplate configuration file. A visualization of the schema gives more insights on required and optional properties can be [found here](https://json-schema.app/view/%23?url=https%3A%2F%2Fraw.githubusercontent.com%2Fnetwerk-digitaal-erfgoed%2Fld-workbench%2Fmain%2Fstatic%2Fld-workbench.schema.json).
 
+### Example YAML File For Configuration Options
+
+```yaml
+name: MyPipeline
+description: Example pipeline configuration
+destination: output/result.ttl
+stages:
+  - name: Stage1
+    iterator:
+      query: "SELECT ?s ?p ?o WHERE { ?s ?p ?o } LIMIT 100"
+      endpoint: "http://example.com/sparql-endpoint"
+      delay: "150 ms"
+    generator:
+      - query: "CONSTRUCT { ?s ?p ?o } WHERE { ?s ?p ?o }"
+        batchSize: 50
+    destination: output/stage1-result.ttl
+  - name: Stage2
+    iterator:
+      query: file://queries/iteratorQuery.rq
+      endpoint: "http://example.com/sparql-endpoint-1"
+      batchSize: 200
+    generator:
+      - query: file://queries/generator1Query.rq
+        endpoint: "http://example.com/sparql-endpoint-1"
+        batchSize: 200
+      - query: file://queries/generator2Query.rq
+        endpoint: "http://example.com/sparql-endpoint-2"
+        batchSize: 100
+    destination: output/stage2-result.ttl
+```
 
+### Configuration Options Table
+
+| Section                          | Variable           | Description                                                                                                         | Required |
+|----------------------------------|--------------------|---------------------------------------------------------------------------------------------------------------------|----------|
+| General Configuration File       | name               | The name of your pipeline, it must be unique over all your configurations.                                          | Yes      |
+|                                  | description        | An optional description for your pipeline.                                                                          | No       |
+|                                  | destination        | The file where the final result of your pipeline is saved.                                                          | No       |
+| Stage                            | name               | The name of your pipeline step, it must be unique within one configuration.                                         | Yes      |
+|                                  | destination        | The file where the results are saved. This is not a required property; if omitted, a temporary file will be created automatically. | No       |
+| Iterator                         | query              | Path (prefixed with "file://") of SPARQL Query `.rq` file or SPARQL Query string that makes the iterator using SPARQL select.                        | Yes      |
+|                                  | endpoint           | The SPARQL endpoint for the iterator. If it starts with "file://", a local RDF file is queried. If omitted, the result of the previous stage is used. | No       |
+|                                  | batchSize          | Overrule the iterator's behavior of fetching 10 results per request, regardless of any limits in your query.       | No       |
+|                                  | delay              | Human-readable time delay for the iterator's SPARQL endpoint requests (e.g., '5ms', '100 milliseconds', '1s').   | No       |
+| Generator                        | query              | Path (prefixed with "file://") of SPARQL Query `.rq` file or SPARQL Query string that makes the generator using SPARQL construct.                    | Yes      |
+|                                  | endpoint           | The SPARQL endpoint for the generator. If it starts with "file://", a local RDF file is queried. If omitted, the endpoint of the Iterator is used. | No       |
+|                                  | batchSize          | Overrule the generator's behavior of fetching results for 10 bindings of $this per request.                          | No       |
 
 ## Installation
 
@@ -47,8 +89,6 @@ A pipeline must have a name, one or more stages, and optionally a description. M
 
    Your workbench is now ready for use.
 
-
-
 ## Usage
 
 Once installed, an example workbench is present that can be run with the following command:
@@ -57,14 +97,13 @@ Once installed, an example workbench is present that can be run with the followi
 npx @netwerkdigitaalergoed/ld-workbench
 ```
 
-
 ### Configuring a workbench pipeline
 
 To keep your workbench workspace clean, create a folder for each pipeline that contains the configuration and the SPARQL Select and Construct queries. Use the `static` directory for this.
 
 Here is an example of how your file structure may look:
 
-```
+```sh
 ld-workbench
 |-- static
 |   |-- my-pipeline
@@ -73,8 +112,6 @@ ld-workbench
 |   |   |-- construct.rq
 ```
 
-
-
 ## Development
 
 For local development, the following command should get you going:
@@ -95,3 +132,21 @@ npm run ld-workbench -- --configDir static/example
 Since this project is written in Typescript, your code needs to be transpiled to Javascript before you can run it (using `npm run build`). With `npm run dev` the transpiler will watch changes in the Typescript code an transpiles on each change.
 
 The configuration of this project is validated and defined by [JSON Schema](https://json-schema.org). The schema is located in `./static/ld-workbench-schema.json`. To create the types from this schema, run `npm run util:json-schema-to-typescript`. This will regenerate `./src/types/LDWorkbenchConfiguration.d.ts`, do not modify this file by hand.
+
+## Publishing to NPM
+
+To trigger a published release event in GitHub Actions, you need to create a release on your GitHub repository. Here are the general steps:
+
+1. Navigate to the [LDWorkbench GitHub repository](https://github.com/netwerk-digitaal-erfgoed/ld-workbench).
+
+2. Click on the "Releases" tab or visit <https://github.com/netwerk-digitaal-erfgoed/ld-workbench/releases>.
+
+3. Click the "Draft a new release" button if you haven't created a release yet.
+
+4. Fill in the necessary information for your release, such as the tag version, release title, and release description.
+
+5. Optionally, attach any assets (e.g. release notes).
+
+6. Click the "Publish release" button.
+
+By publishing the release, you will trigger the "release" event in GitHub Actions, which can then execute the workflow specified [here](https://github.com/netwerk-digitaal-erfgoed/ld-workbench/blob/main/.github/workflows/npmjs.yml).
diff --git a/src/lib/Iterator.class.ts b/src/lib/Iterator.class.ts
@@ -30,14 +30,14 @@ class Iterator extends EventEmitter {
   private readonly delay: number | undefined
   private source: string = "";
   private $offset = 0;
-  private totalResults = 0;
+  public totalResults = 0;
 
   constructor(stage: Stage) {
     super();
     this.query = getSPARQLQuery(stage.configuration.iterator.query, "select");
     this.query.limit =
-      this.query.limit ??
       stage.configuration.iterator.batchSize ??
+      this.query.limit ??
       DEFAULT_LIMIT;
     this.endpoint = getEndpoint(stage);
     this.engine = getEngine(this.endpoint);

diff --git a/src/lib/Pipeline.class.ts b/src/lib/Pipeline.class.ts
@@ -5,7 +5,6 @@ import chalk from "chalk";
 import Stage from "./Stage.class.js";
 import formatDuration from "../utils/formatDuration.js";
 import { millify } from "millify";
-import { setTimeout } from "node:timers/promises";
 import File from "./File.class.js";
 import path from "node:path";
 import * as fs from "node:fs";
@@ -183,9 +182,9 @@ class Pipeline {
     let iterationsProcessed = 0
     if (!(this.opts?.silent === true)) spinner.start();
     await new Promise<void>((resolve, reject) => {
-      stage.on("iteratorResult", (_$this) => {
+      stage.on("iteratorResult", (_$this, quadsGenerated) => {
         iterationsProcessed++
-        if (!(this.opts?.silent === true)) spinner.text = `Running ${stage.name}:\n\n  Elements processed: ${millify(iterationsProcessed)} \n  Duration: ${formatDuration(startTime, performance.now())} `;
+        if (!(this.opts?.silent === true)) spinner.text = `Running ${stage.name}:\n\n  Processed elements: ${millify(iterationsProcessed)}\n  Generated quads: ${millify(quadsGenerated)}\n  Duration: ${formatDuration(startTime, performance.now())} `;
       });
       stage.on("error", (e) => {
         spinner.fail();
@@ -213,7 +212,6 @@ class Pipeline {
 
     if (this.stageNames.length !== 0) return this.runRecursive();
     try {
-      await setTimeout(3000)
       await this.writeResult();
     } catch (e) {
       throw new Error("Pipeline failed: " + (e as Error).message);
@@ -227,7 +225,6 @@ class Pipeline {
           )}" was completed in ${formatDuration(this.startTime, performance.now())}`
         )
       );
-    process.exit(0)
   }
 
   private async writeResult(): Promise<void> {

diff --git a/src/lib/Stage.class.ts b/src/lib/Stage.class.ts
@@ -13,27 +13,25 @@ import type { WriteStream } from 'node:fs';
 declare interface Stage {
   on(event: "generatorResult", listener: (count: number) => void): this;
   on(event: "end", listener: (iteratorCount: number, statements: number) => void): this;
-  on(event: "iteratorResult", listener: ($this: NamedNode) => void): this;
+  on(event: "iteratorResult", listener: ($this: NamedNode, quadsGenerated: number) => void): this;
   on(event: "error", listener: (e: Error) => void): this;
 
   emit(event: "generatorResult", count: number): boolean;
   emit(event: "end", iteratorCount: number, statements: number): boolean;
-  emit(event: "iteratorResult", $this: NamedNode): boolean;
+  emit(event: "iteratorResult", $this: NamedNode, quadsGenerated: number): boolean;
   emit(event: "error", e: Error): boolean;
 }
 
 class Stage extends EventEmitter {
   public destination: () => WriteStream
   public iterator: Iterator
   public generators: Generator[] = []
-  private totalProcessed: number
 
   public constructor(
     public readonly pipeline: Pipeline,
     public readonly configuration: LDWorkbenchConfiguration['stages'][0]
   ) {
     super()
-    this.totalProcessed = 0
     try {
       this.iterator = new Iterator(this)
     } catch(e) {
@@ -59,35 +57,50 @@ class Stage extends EventEmitter {
   public get name(): string {
     return this.configuration.name
   }
-
+  
   public run(): void {
     const writer = new Writer(this.destination(), { end: false, format: 'N-Triples' });
     let quadCount = 0;
 
-    this.generators.forEach(generator => {
+    const generatorProcessedCounts = new Map<number, number>();
+    let generatorsFinished = 0;
+    let quadsGenerated = 0;
+
+    const checkEnd = (iterationsIncoming: number, statements: number): void => {
+      // Check if all generators have processed all iterations
+      if (generatorsFinished === this.configuration.generator.length) {
+        this.emit('end', iterationsIncoming, statements);
+      }
+    };
+
+    this.generators.forEach((generator, index) => {
+      generatorProcessedCounts.set(index, 0);
+
       generator.on('data', (quad) => {
-          writer.addQuad(quad);
-          quadCount++;
-          this.emit('generatorResult', quadCount);
+        quadsGenerated++
+        writer.addQuad(quad);
+        quadCount++;
+        this.emit('generatorResult', quadCount);
       });
 
       generator.on('end', (iterationsIncoming, statements, processed) => {
-        this.totalProcessed += processed
-        if (this.totalProcessed >= (iterationsIncoming * this.configuration.generator.length)){
-          this.emit('end', iterationsIncoming, statements);
+        generatorProcessedCounts.set(index, generatorProcessedCounts.get(index)! + processed);
+        if (generatorProcessedCounts.get(index)! >= iterationsIncoming) {
+          generatorsFinished++;
         }
-    });
+        checkEnd(iterationsIncoming, statements);
+      });
 
-      generator.on('error', e => {
-        this.emit('error', e)
-      })
+      generator.on('error', (e) => {
+        this.emit('error', e);
+      });
     });
 
     this.iterator.on('data', ($this) => {
       this.generators.forEach(generator => {
         generator.run($this);
       });
-      this.emit('iteratorResult', $this);
+      this.emit('iteratorResult', $this, quadsGenerated);
   });
 
   this.iterator.on('error', e => {

diff --git a/src/lib/tests/Iterator.class.test.ts b/src/lib/tests/Iterator.class.test.ts
@@ -66,7 +66,6 @@ describe('Iterator Class', () => {
             expect(iterator).to.have.property('totalResults', 0);
         });
     });
-    // BUG when both the generator and iterator tests are running, it seems the iterator will never terminate
     describe.skip('run', () => {
         it('should emit "data" and "end" events with the correct $this and numResults', async () => {
             const configuration: LDWorkbenchConfiguration = {

diff --git a/src/lib/tests/Pipeline.class.test.ts b/src/lib/tests/Pipeline.class.test.ts
@@ -9,8 +9,6 @@ import removeDirectory from "../../utils/removeDir.js";
 chai.use(chaiAsPromised)
 const expect = chai.expect
 
-
-// BUG currently the pipeline class fails with no such file or directory, open 'pipelines/data/example-pipeline-batch/stage-1.nt and done() called multiple times in test <Pipeline Class run should run the pipeline correctly> of file /Users/work/triply/NDE-LDWorkbench/ld-workbench/dist/lib/tests/Pipeline.class.test.js
 describe('Pipeline Class', () => {
     const dataDirectoryPath = path.join('pipelines', 'data');
 

diff --git a/src/lib/tests/Stage.class.test.ts b/src/lib/tests/Stage.class.test.ts
@@ -148,7 +148,6 @@ describe('Stage Class', () => {
         });
     });
 
-    // BUG throws error when in combined test on stage's Iterator, when set to only it will pass.
     describe.skip('run', () => {
         it('should run the stage correctly', async function () {
             const configuration: LDWorkbenchConfiguration = {

diff --git a/src/main.tsx b/src/main.tsx
@@ -61,6 +61,7 @@ async function main(): Promise<void> {
 
   try {
     await pipeline.run();
+    process.exit(0)
   } catch (e) {
     error(`Error in pipeline ${chalk.italic(configuration.name)}`, 5, e as Error);
   }