fix: Support BGPs with variables in SPARQL UPDATE queries

package.json

@@ -75,6 +75,7 @@
     "templates"
   ],
   "dependencies": {
+    "@comunica/actor-init-sparql": "^1.21.3",
     "@rdfjs/data-model": "^1.2.0",
     "@rdfjs/fetch": "^2.1.0",
     "@solid/identity-token-verifier": "^0.8.0",
@@ -119,7 +120,6 @@
     "punycode": "^2.1.1",
     "rdf-parse": "^1.7.0",
     "rdf-serialize": "^1.1.0",
-    "rdf-terms": "^1.5.1",
     "redis": "^3.0.2",
     "redlock": "^4.2.0",
     "sparqlalgebrajs": "^3.0.0",

src/storage/patch/SparqlUpdatePatchHandler.ts

@@ -1,8 +1,10 @@
 import type { Readable } from 'stream';
+import type { ActorInitSparql } from '@comunica/actor-init-sparql';
+import { newEngine } from '@comunica/actor-init-sparql';
+import type { IQueryResultUpdate } from '@comunica/actor-init-sparql/lib/ActorInitSparql-browser';
 import { defaultGraph } from '@rdfjs/data-model';
 import { Store } from 'n3';
 import type { BaseQuad } from 'rdf-js';
-import { someTerms } from 'rdf-terms';
 import { Algebra } from 'sparqlalgebrajs';
 import type { Patch } from '../../ldp/http/Patch';
 import type { SparqlUpdatePatch } from '../../ldp/http/SparqlUpdatePatch';
@@ -13,6 +15,7 @@ import type { ResourceIdentifier } from '../../ldp/representation/ResourceIdenti
 import { getLoggerFor } from '../../logging/LogUtil';
 import { INTERNAL_QUADS } from '../../util/ContentTypes';
 import { NotImplementedHttpError } from '../../util/errors/NotImplementedHttpError';
+import { readableToString } from '../../util/StreamUtil';
 import type { RepresentationConverter } from '../conversion/RepresentationConverter';
 import { ConvertingPatchHandler } from './ConvertingPatchHandler';
 import type { PatchHandlerArgs } from './PatchHandler';
@@ -25,8 +28,11 @@ import type { PatchHandlerArgs } from './PatchHandler';
 export class SparqlUpdatePatchHandler extends ConvertingPatchHandler {
   protected readonly logger = getLoggerFor(this);
 
+  private readonly engine: ActorInitSparql;
+
   public constructor(converter: RepresentationConverter, defaultType = 'text/turtle') {
     super(converter, INTERNAL_QUADS, defaultType);
+    this.engine = newEngine();
   }
 
   public async canHandle({ patch }: PatchHandlerArgs): Promise<void> {
@@ -63,14 +69,6 @@ export class SparqlUpdatePatchHandler extends ConvertingPatchHandler {
     return op.type === Algebra.types.COMPOSITE_UPDATE;
   }
 
-  private isBasicGraphPatternWithoutVariables(op: Algebra.Operation): op is Algebra.Bgp {
-    if (op.type !== Algebra.types.BGP) {
-      return false;
-    }
-    return !(op.patterns as BaseQuad[]).some((pattern): boolean =>
-      someTerms(pattern, (term): boolean => term.termType === 'Variable'));
-  }
-
   /**
    * Checks if the input operation is of a supported type (DELETE/INSERT or composite of those)
    */
@@ -101,9 +99,9 @@ export class SparqlUpdatePatchHandler extends ConvertingPatchHandler {
       this.logger.warn('GRAPH statement in INSERT clause');
       throw new NotImplementedHttpError('GRAPH statements are not supported');
     }
-    if (!(typeof op.where === 'undefined' || this.isBasicGraphPatternWithoutVariables(op.where))) {
-      this.logger.warn('WHERE statements are not supported');
-      throw new NotImplementedHttpError('WHERE statements are not supported');
+    if (!(typeof op.where === 'undefined' || op.where.type === Algebra.types.BGP)) {
+      this.logger.warn('Non-BGP WHERE statements are not supported');
+      throw new NotImplementedHttpError('Non-BGP WHERE statements are not supported');
     }
   }
 
@@ -136,41 +134,14 @@ export class SparqlUpdatePatchHandler extends ConvertingPatchHandler {
       metadata = new RepresentationMetadata(identifier, INTERNAL_QUADS);
     }
 
-    this.applyOperation(result, (patch as SparqlUpdatePatch).algebra);
+    // Run the query through Comunica
+    const sparql = await readableToString(patch.data);
+    const query = await this.engine.query(sparql,
+      { sources: [ result ], baseIRI: identifier.path }) as IQueryResultUpdate;
+    await query.updateResult;
+
     this.logger.debug(`${result.size} quads will be stored to ${identifier.path}.`);
 
     return new BasicRepresentation(result.match() as Readable, metadata);
   }
-
-  /**
-   * Apply the given algebra update operation to the store of quads.
-   */
-  private applyOperation(store: Store<BaseQuad>, op: Algebra.Operation): void {
-    if (this.isDeleteInsert(op)) {
-      this.applyDeleteInsert(store, op);
-    // Only other options is Composite after passing `validateUpdate`
-    } else {
-      this.applyComposite(store, op as Algebra.CompositeUpdate);
-    }
-  }
-
-  /**
-   * Apply the given composite update operation to the store of quads.
-   */
-  private applyComposite(store: Store<BaseQuad>, op: Algebra.CompositeUpdate): void {
-    for (const update of op.updates) {
-      this.applyOperation(store, update);
-    }
-  }
-
-  /**
-   * Apply the given DELETE/INSERT update operation to the store of quads.
-   */
-  private applyDeleteInsert(store: Store<BaseQuad>, op: Algebra.DeleteInsert): void {
-    const deletes = op.delete ?? [];
-    const inserts = op.insert ?? [];
-    store.removeQuads(deletes);
-    store.addQuads(inserts);
-    this.logger.debug(`Removed ${deletes.length} and added ${inserts.length} quads.`);
-  }
 }

test/unit/storage/patch/SparqlUpdatePatchHandler.test.ts

@@ -13,6 +13,7 @@ import type { ResourceStore } from '../../../../src/storage/ResourceStore';
 import { INTERNAL_QUADS } from '../../../../src/util/ContentTypes';
 import { NotFoundHttpError } from '../../../../src/util/errors/NotFoundHttpError';
 import { NotImplementedHttpError } from '../../../../src/util/errors/NotImplementedHttpError';
+import { guardedStreamFrom } from '../../../../src/util/StreamUtil';
 
 describe('A SparqlUpdatePatchHandler', (): void => {
   let converter: jest.Mocked<RepresentationConverter>;
@@ -75,11 +76,16 @@ describe('A SparqlUpdatePatchHandler', (): void => {
   }
 
   async function handle(query: string): Promise<void> {
-    const sparqlPrefix = 'prefix : <http://test.com/>\n';
+    const prefixedQuery = `prefix : <http://test.com/>\n${query}`;
     await handler.handle({
       source,
       identifier,
-      patch: { algebra: translate(sparqlPrefix.concat(query), { quads: true }) } as SparqlUpdatePatch,
+      patch: {
+        algebra: translate(prefixedQuery, { quads: true }),
+        data: guardedStreamFrom(prefixedQuery),
+        metadata: new RepresentationMetadata(),
+        binary: true,
+      } as SparqlUpdatePatch,
     });
   }
 
@@ -124,6 +130,16 @@ describe('A SparqlUpdatePatchHandler', (): void => {
     )).toBe(true);
   });
 
+  it('handles DELETE WHERE updates with variables.', async(): Promise<void> => {
+    const query = 'DELETE WHERE { :startS1 :startP1 ?o }';
+    await handle(query);
+    expect(await basicChecks(
+      [ quad(namedNode('http://test.com/startS2'),
+        namedNode('http://test.com/startP2'),
+        namedNode('http://test.com/startO2')) ],
+    )).toBe(true);
+  });
+
   it('handles DELETE/INSERT updates with empty WHERE.', async(): Promise<void> => {
     const query = 'DELETE { :startS1 :startP1 :startO1 } INSERT { :s1 :p1 :o1 . } WHERE {}';
     await handle(query);
@@ -178,8 +194,8 @@ describe('A SparqlUpdatePatchHandler', (): void => {
     await expect(handle(query)).rejects.toThrow(NotImplementedHttpError);
   });
 
-  it('rejects DELETE/INSERT updates with a non-empty WHERE.', async(): Promise<void> => {
-    const query = 'DELETE { :s1 :p1 :o1 } INSERT { :s1 :p1 :o1 } WHERE { ?s ?p ?o }';
+  it('rejects DELETE/INSERT updates with non-BGP WHERE.', async(): Promise<void> => {
+    const query = 'DELETE { :s1 :p1 :o1 } INSERT { :s1 :p1 :o1 } WHERE { ?s ?p ?o. FILTER (?o > 5) }';
     await expect(handle(query)).rejects.toThrow(NotImplementedHttpError);
   });
 
@@ -188,11 +204,6 @@ describe('A SparqlUpdatePatchHandler', (): void => {
     await expect(handle(query)).rejects.toThrow(NotImplementedHttpError);
   });
 
-  it('rejects DELETE WHERE updates with variables.', async(): Promise<void> => {
-    const query = 'DELETE WHERE { ?v :startP1 :startO1 }';
-    await expect(handle(query)).rejects.toThrow(NotImplementedHttpError);
-  });
-
   it('rejects non-DELETE/INSERT updates.', async(): Promise<void> => {
     const query = 'MOVE DEFAULT TO GRAPH :newGraph';
     await expect(handle(query)).rejects.toThrow(NotImplementedHttpError);