add in-memory SPARQL

.changeset/six-flowers-look.md

@@ -0,0 +1,5 @@
+---
+"barnard59-sparql": minor
+---
+
+Add In-Memory SPARQL operations (closes #255)

packages/sparql/inMemory.js

@@ -0,0 +1,22 @@
+import { Transform } from 'readable-stream'
+import oxigraph from 'oxigraph'
+
+export const update = sparql =>
+  new Transform({
+    objectMode: true,
+    transform: (chunk, encoding, callback) => {
+      const store = new oxigraph.Store([...chunk])
+      store.update(sparql)
+      callback(null, store.match())
+    },
+  })
+
+export const query = sparql =>
+  new Transform({
+    objectMode: true,
+    transform: (chunk, encoding, callback) => {
+      const store = new oxigraph.Store([...chunk])
+      const result = store.query(sparql)
+      callback(null, result)
+    },
+  })

packages/sparql/manifest.ttl

@@ -4,7 +4,7 @@
 @prefix rdfs: <http://www.w3.org/2000/01/rdf-schema#> .
 
 <construct> a p:Operation, p:ReadableObjectMode;
-  rdfs:label "SPARQL Select";
+  rdfs:label "SPARQL Construct";
   rdfs:comment "Runs the given CONSTRUCT query against the given endpoint parses the result.";
   code:implementedBy [ a code:EcmaScriptModule;
     code:link <node:barnard59-sparql/construct.js#default>
@@ -16,3 +16,17 @@
   code:implementedBy [ a code:EcmaScriptModule;
     code:link <node:barnard59-sparql/select.js#default>
   ].
+
+<in-memory/update> a p:Operation, p:ReadableObjectMode;
+  rdfs:label "SPARQL in-memory Update";
+  rdfs:comment "Runs the given DELETE/INSERT command against each input chunk.";
+  code:implementedBy [ a code:EcmaScriptModule;
+    code:link <node:barnard59-sparql/inMemory.js#update>
+  ].
+
+<in-memory/query> a p:Operation, p:ReadableObjectMode;
+  rdfs:label "SPARQL in-memory Query";
+  rdfs:comment "Runs the given query against each input chunk.";
+  code:implementedBy [ a code:EcmaScriptModule;
+    code:link <node:barnard59-sparql/inMemory.js#query>
+  ].

packages/sparql/package.json

@@ -19,13 +19,15 @@
   "homepage": "https://github.com/zazuko/barnard59-sparql",
   "dependencies": {
     "duplex-to": "^1.0.1",
+    "oxigraph": "^0.4.0-alpha.2",
+    "readable-stream": "^4.5.2",
     "sparql-http-client": "^2.4.0"
   },
   "devDependencies": {
     "@tpluscode/rdf-string": "^1.1.2",
     "@zazuko/env": "^2",
     "get-stream": "^6.0.0",
-    "isstream": "^0.1.2",
+    "is-stream": "^3.0.0",
     "nock": "^13.1.1"
   },
   "engines": {

packages/sparql/test/construct.test.js

@@ -1,6 +1,6 @@
 import { strictEqual } from 'assert'
 import getStream from 'get-stream'
-import { isReadable, isWritable } from 'isstream'
+import { isReadableStream, isWritableStream } from 'is-stream'
 import nock from 'nock'
 import rdf from '@zazuko/env'
 import { turtle } from '@tpluscode/rdf-string'
@@ -22,8 +22,8 @@ describe('construct', () => {
 
     const result = await construct({ endpoint, query })
 
-    strictEqual(isReadable(result), true)
-    strictEqual(isWritable(result), false)
+    strictEqual(isReadableStream(result), true)
+    strictEqual(isWritableStream(result), false)
   })
 
   it('should send a GET request', async () => {

packages/sparql/test/inMemory.test.js

@@ -0,0 +1,75 @@
+import { ok, strictEqual } from 'assert'
+import getStream from 'get-stream'
+import { isReadableStream, isWritableStream } from 'is-stream'
+import { Readable } from 'readable-stream'
+import rdf from '@zazuko/env'
+import { query, update } from '../inMemory.js'
+import * as ns from './support/namespaces.js'
+
+describe('query', () => {
+  it('should be a function', () => {
+    strictEqual(typeof query, 'function')
+  })
+
+  it('should return a readable and witable stream', () => {
+    const construct = 'CONSTRUCT { ?s ?p ?o } WHERE { ?s ?p ?o }'
+
+    const result = query(construct)
+
+    strictEqual(isReadableStream(result), true)
+    strictEqual(isWritableStream(result), true)
+  })
+
+  it('should CONSTRUCT quads', async () => {
+    const chunk1 = [
+      rdf.quad(ns.ex.s1, ns.ex.p, ns.ex.o1),
+      rdf.quad(ns.ex.s2, ns.ex.p, ns.ex.s2),
+    ]
+    const chunk2 = [
+      rdf.quad(ns.ex.s3, ns.ex.p, ns.ex.o3),
+    ]
+
+    const construct = query('CONSTRUCT { ?s ?p "ok" } WHERE { ?s ?p ?s }')
+    const pipeline = Readable.from([chunk1, chunk2]).pipe(construct)
+    const result = await getStream.array(pipeline)
+    const dataset = rdf.dataset(result.flat())
+
+    strictEqual(result.length, 2)
+    strictEqual(dataset.size, 1)
+    ok(dataset.has(rdf.quad(ns.ex.s2, ns.ex.p, rdf.literal('ok'))))
+  })
+})
+
+describe('update', () => {
+  it('should be a function', () => {
+    strictEqual(typeof update, 'function')
+  })
+
+  it('should return a readable and witable stream', () => {
+    const construct = 'DELETE { ?s ?p ?o } WHERE { ?s ?p ?o }'
+
+    const result = update(construct)
+
+    strictEqual(isReadableStream(result), true)
+    strictEqual(isWritableStream(result), true)
+  })
+  it('should UPDATE quads', async () => {
+    const chunk1 = [
+      rdf.quad(ns.ex.s1, ns.ex.p, ns.ex.o1),
+      rdf.quad(ns.ex.s2, ns.ex.p, ns.ex.s2),
+    ]
+    const chunk2 = [
+      rdf.quad(ns.ex.s3, ns.ex.p, ns.ex.o3),
+    ]
+
+    const command = update('DELETE { ?s ?p ?s } WHERE { ?s ?p ?s }')
+    const pipeline = Readable.from([chunk1, chunk2]).pipe(command)
+    const result = await getStream.array(pipeline)
+
+    strictEqual(result.length, 2)
+    const dataset = rdf.dataset(result.flat())
+    strictEqual(dataset.size, 2)
+    ok(dataset.has(rdf.quad(ns.ex.s1, ns.ex.p, ns.ex.o1)))
+    ok(dataset.has(rdf.quad(ns.ex.s3, ns.ex.p, ns.ex.o3)))
+  })
+})

packages/sparql/test/select.test.js

@@ -1,6 +1,6 @@
 import { strictEqual } from 'assert'
 import getStream from 'get-stream'
-import { isReadable, isWritable } from 'isstream'
+import { isReadableStream, isWritableStream } from 'is-stream'
 import nock from 'nock'
 import select from '../select.js'
 
@@ -19,8 +19,8 @@ describe('select', () => {
 
     const result = await select({ endpoint, query })
 
-    strictEqual(isReadable(result), true)
-    strictEqual(isWritable(result), false)
+    strictEqual(isReadableStream(result), true)
+    strictEqual(isWritableStream(result), false)
   })
 
   it('should send a GET request', async () => {