Enable incremental metadata updates

This refactors internal metadata handling so that it can be incrementally updated.

Metadata objects now have a `state` field, which represents their validation state,
which can be invalidated, and be listened to.
This allows actors to listen to metadata invalidations and act upon changes.

This commit also adds a metadata accumulation bus that
abstracts the logic for merging metadata from multiple sources that
existed before in the federated actor.
This will enable this logic to be used in other places, and for other
actors to manage how metadata is merged.

Closes #1156
Closes #1180

May be related to comunica/comunica-feature-link-traversal#102

.componentsjs-generator-config.json

@@ -53,6 +53,7 @@
     "IQueryableResultVoid",
     "PathVariableObjectIterator",
     "ClosableTransformIterator",
+    "MetadataValidationState",
 
     "RDF.Stream",
     "Omit",

engines/config-query-sparql/config/config-default.json

@@ -32,6 +32,8 @@
     "ccqs:config/rdf-join-selectivity/mediators.json",
     "ccqs:config/rdf-metadata/actors.json",
     "ccqs:config/rdf-metadata/mediators.json",
+    "ccqs:config/rdf-metadata-accumulate/actors.json",
+    "ccqs:config/rdf-metadata-accumulate/mediators.json",
     "ccqs:config/rdf-metadata-extract/actors.json",
     "ccqs:config/rdf-metadata-extract/mediators.json",
     "ccqs:config/rdf-parse/actors.json",

engines/config-query-sparql/config/config-rdfjs.json

@@ -28,6 +28,9 @@
     "ccqs:config/rdf-join-selectivity/actors.json",
     "ccqs:config/rdf-join-selectivity/mediators.json",
 
+    "ccqs:config/rdf-metadata-accumulate/actors.json",
+    "ccqs:config/rdf-metadata-accumulate/mediators.json",
+
     "ccqs:config/rdf-resolve-quad-pattern/actors-rdfjs.json",
     "ccqs:config/rdf-resolve-quad-pattern/mediators.json",
     "ccqs:config/rdf-serialize/actors.json",

engines/config-query-sparql/config/rdf-metadata-accumulate/actors.json

@@ -0,0 +1,30 @@
+{
+  "@context": [
+    "https://linkedsoftwaredependencies.org/bundles/npm/@comunica/runner/^2.0.0/components/context.jsonld",
+
+    "https://linkedsoftwaredependencies.org/bundles/npm/@comunica/actor-rdf-metadata-accumulate-cancontainundefs/^2.0.0/components/context.jsonld",
+    "https://linkedsoftwaredependencies.org/bundles/npm/@comunica/actor-rdf-metadata-accumulate-cardinality/^2.0.0/components/context.jsonld",
+    "https://linkedsoftwaredependencies.org/bundles/npm/@comunica/actor-rdf-metadata-accumulate-pagesize/^2.0.0/components/context.jsonld",
+    "https://linkedsoftwaredependencies.org/bundles/npm/@comunica/actor-rdf-metadata-accumulate-requesttime/^2.0.0/components/context.jsonld"
+  ],
+  "@id": "urn:comunica:default:Runner",
+  "@type": "Runner",
+  "actors": [
+    {
+      "@id": "urn:comunica:default:rdf-metadata-accumulate/actors#cancontainundefs",
+      "@type": "ActorRdfMetadataAccumulateCanContainUndefs"
+    },
+    {
+      "@id": "urn:comunica:default:rdf-metadata-accumulate/actors#cardinality",
+      "@type": "ActorRdfMetadataAccumulateCardinality"
+    },
+    {
+      "@id": "urn:comunica:default:rdf-metadata-accumulate/actors#pagesize",
+      "@type": "ActorRdfMetadataAccumulatePageSize"
+    },
+    {
+      "@id": "urn:comunica:default:rdf-metadata-accumulate/actors#requesttime",
+      "@type": "ActorRdfMetadataAccumulateRequestTime"
+    }
+  ]
+}

engines/config-query-sparql/config/rdf-metadata-accumulate/mediators.json

@@ -0,0 +1,10 @@
+{
+  "@context": [
+    "https://linkedsoftwaredependencies.org/bundles/npm/@comunica/bus-rdf-metadata-accumulate/^2.0.0/components/context.jsonld",
+    "https://linkedsoftwaredependencies.org/bundles/npm/@comunica/mediator-combine-union/^2.0.0/components/context.jsonld"
+  ],
+  "@id": "urn:comunica:default:rdf-metadata-accumulate/mediators#main",
+  "@type": "MediatorCombineUnion",
+  "bus": { "@id": "ActorRdfMetadataAccumulate:_default_bus" },
+  "field": "metadata"
+}

engines/config-query-sparql/config/rdf-resolve-quad-pattern/actors/federated.json

@@ -10,7 +10,8 @@
     {
       "@id": "urn:comunica:default:rdf-resolve-quad-pattern/actors#federated",
       "@type": "ActorRdfResolveQuadPatternFederated",
-      "mediatorResolveQuadPattern": { "@id": "urn:comunica:default:rdf-resolve-quad-pattern/mediators#main" }
+      "mediatorResolveQuadPattern": { "@id": "urn:comunica:default:rdf-resolve-quad-pattern/mediators#main" },
+      "mediatorRdfMetadataAccumulate": { "@id": "urn:comunica:default:rdf-metadata-accumulate/mediators#main" }
     }
   ]
 }

engines/config-query-sparql/config/rdf-resolve-quad-pattern/actors/hypermedia.json

@@ -13,6 +13,7 @@
       "mediatorDereferenceRdf": { "@id": "urn:comunica:default:dereference-rdf/mediators#main" },
       "mediatorMetadata": { "@id": "urn:comunica:default:rdf-metadata/mediators#main" },
       "mediatorMetadataExtract": { "@id": "urn:comunica:default:rdf-metadata-extract/mediators#main" },
+      "mediatorMetadataAccumulate": { "@id": "urn:comunica:default:rdf-metadata-accumulate/mediators#main" },
       "mediatorRdfResolveHypermedia": { "@id": "urn:comunica:default:rdf-resolve-hypermedia/mediators#main" },
       "mediatorRdfResolveHypermediaLinks": { "@id": "urn:comunica:default:rdf-resolve-hypermedia-links/mediators#main" },
       "mediatorRdfResolveHypermediaLinksQueue": { "@id": "urn:comunica:default:rdf-resolve-hypermedia-links-queue/mediators#main" }

engines/query-sparql-file/package.json

@@ -125,6 +125,10 @@
     "@comunica/actor-rdf-join-optional-bind": "^2.6.8",
     "@comunica/actor-rdf-join-optional-nestedloop": "^2.6.8",
     "@comunica/actor-rdf-join-selectivity-variable-counting": "^2.6.8",
+    "@comunica/actor-rdf-metadata-accumulate-cancontainundefs": "^2.5.0",
+    "@comunica/actor-rdf-metadata-accumulate-cardinality": "^2.5.0",
+    "@comunica/actor-rdf-metadata-accumulate-pagesize": "^2.5.0",
+    "@comunica/actor-rdf-metadata-accumulate-requesttime": "^2.5.0",
     "@comunica/actor-rdf-metadata-all": "^2.6.8",
     "@comunica/actor-rdf-metadata-extract-allow-http-methods": "^2.6.8",
     "@comunica/actor-rdf-metadata-extract-hydra-controls": "^2.6.8",

engines/query-sparql-rdfjs/package.json

@@ -110,6 +110,10 @@
     "@comunica/actor-rdf-join-optional-bind": "^2.6.8",
     "@comunica/actor-rdf-join-optional-nestedloop": "^2.6.8",
     "@comunica/actor-rdf-join-selectivity-variable-counting": "^2.6.8",
+    "@comunica/actor-rdf-metadata-accumulate-cancontainundefs": "^2.5.0",
+    "@comunica/actor-rdf-metadata-accumulate-cardinality": "^2.5.0",
+    "@comunica/actor-rdf-metadata-accumulate-pagesize": "^2.5.0",
+    "@comunica/actor-rdf-metadata-accumulate-requesttime": "^2.5.0",
     "@comunica/actor-rdf-resolve-quad-pattern-federated": "^2.6.8",
     "@comunica/actor-rdf-resolve-quad-pattern-rdfjs-source": "^2.6.8",
     "@comunica/actor-rdf-serialize-jsonld": "^2.6.8",

engines/query-sparql/package.json

@@ -123,6 +123,10 @@
     "@comunica/actor-rdf-join-optional-bind": "^2.6.8",
     "@comunica/actor-rdf-join-optional-nestedloop": "^2.6.8",
     "@comunica/actor-rdf-join-selectivity-variable-counting": "^2.6.8",
+    "@comunica/actor-rdf-metadata-accumulate-cancontainundefs": "^2.5.0",
+    "@comunica/actor-rdf-metadata-accumulate-cardinality": "^2.5.0",
+    "@comunica/actor-rdf-metadata-accumulate-pagesize": "^2.5.0",
+    "@comunica/actor-rdf-metadata-accumulate-requesttime": "^2.5.0",
     "@comunica/actor-rdf-metadata-all": "^2.6.8",
     "@comunica/actor-rdf-metadata-extract-allow-http-methods": "^2.6.8",
     "@comunica/actor-rdf-metadata-extract-hydra-controls": "^2.6.8",

lerna.json

@@ -8,7 +8,8 @@
       "message": "Bump to release version %s",
       "allowBranch": [
         "master",
-        "next/major"
+        "next/major",
+        "feature/adaptive-join"
       ],
       "npmClient": "npm"
     }

packages/actor-init-query/test/QueryEngineBase-test.ts

@@ -2,6 +2,7 @@ import { Readable, Transform } from 'stream';
 import { BindingsFactory } from '@comunica/bindings-factory';
 import { KeysInitQuery } from '@comunica/context-entries';
 import { Bus, ActionContext, ActionContextKey } from '@comunica/core';
+import { MetadataValidationState } from '@comunica/metadata';
 import type {
   IPhysicalQueryPlanLogger,
   IActionContext, QueryStringContext, IQueryBindingsEnhanced, IQueryQuadsEnhanced,
@@ -551,6 +552,7 @@ describe('QueryEngineBase', () => {
           ]),
         ]),
         metadata: async() => ({
+          state: new MetadataValidationState(),
           cardinality: { type: 'estimate', value: 1 },
           canContainUndefs: false,
           variables: [ DF.variable('a') ],
@@ -565,6 +567,7 @@ describe('QueryEngineBase', () => {
         ]),
       ]);
       expect(await final.metadata()).toEqual({
+        state: expect.any(MetadataValidationState),
         cardinality: { type: 'estimate', value: 1 },
         canContainUndefs: false,
         variables: [ DF.variable('a') ],
@@ -579,6 +582,7 @@ describe('QueryEngineBase', () => {
           DF.quad(DF.namedNode('ex:a'), DF.namedNode('ex:a'), DF.namedNode('ex:a')),
         ]),
         metadata: async() => ({
+          state: new MetadataValidationState(),
           cardinality: { type: 'estimate', value: 1 },
           canContainUndefs: false,
           variables: [ DF.variable('a') ],
@@ -591,6 +595,7 @@ describe('QueryEngineBase', () => {
         DF.quad(DF.namedNode('ex:a'), DF.namedNode('ex:a'), DF.namedNode('ex:a')),
       ]);
       expect(await final.metadata()).toEqual({
+        state: expect.any(MetadataValidationState),
         cardinality: { type: 'estimate', value: 1 },
         canContainUndefs: false,
         variables: [ DF.variable('a') ],

packages/actor-query-operation-describe-subject/test/ActorQueryOperationDescribeSubject-test.ts

@@ -1,5 +1,6 @@
 import { ActorQueryOperation } from '@comunica/bus-query-operation';
 import { ActionContext, Bus } from '@comunica/core';
+import { MetadataValidationState } from '@comunica/metadata';
 import type { IQueryOperationResultQuads } from '@comunica/types';
 import type * as RDF from '@rdfjs/types';
 import arrayifyStream from 'arrayify-stream';
@@ -21,6 +22,7 @@ describe('ActorQueryOperationDescribeSubject', () => {
           const patterns = [ ...arg.operation.input.input[0].patterns, ...arg.operation.input.input[1].patterns ];
           return {
             metadata: () => Promise.resolve({
+              state: new MetadataValidationState(),
               cardinality: {
                 type: 'estimate',
                 value: arg.operation.input.input[0].patterns.length + arg.operation.input.input[1].patterns.length,
@@ -36,6 +38,7 @@ describe('ActorQueryOperationDescribeSubject', () => {
         }
         return {
           metadata: () => Promise.resolve({
+            state: new MetadataValidationState(),
             cardinality: { type: 'estimate', value: arg.operation.input.patterns.length },
           }),
           quadStream: new ArrayIterator(arg.operation.input.patterns.map(
@@ -94,7 +97,9 @@ describe('ActorQueryOperationDescribeSubject', () => {
       };
       return actor.run(op).then(async(output: IQueryOperationResultQuads) => {
         expect(await output.metadata())
-          .toEqual({ cardinality: { type: 'estimate', value: 2 }, canContainUndefs: false });
+          .toEqual({ state: expect.any(MetadataValidationState),
+            cardinality: { type: 'estimate', value: 2 },
+            canContainUndefs: false });
         expect(output.type).toEqual('quads');
         expect(await arrayifyStream(output.quadStream)).toEqual([
           DF.quad(DF.namedNode('a'), DF.namedNode('__predicate'), DF.namedNode('__object')),
@@ -114,7 +119,9 @@ describe('ActorQueryOperationDescribeSubject', () => {
       };
       return actor.run(op).then(async(output: IQueryOperationResultQuads) => {
         expect(await output.metadata())
-          .toEqual({ cardinality: { type: 'estimate', value: 3 }, canContainUndefs: false });
+          .toEqual({ state: expect.any(MetadataValidationState),
+            cardinality: { type: 'estimate', value: 3 },
+            canContainUndefs: false });
         expect(output.type).toEqual('quads');
         expect(await arrayifyStream(output.quadStream)).toEqual([
           DF.quad(DF.namedNode('a'), DF.namedNode('b'), DF.namedNode('dummy')),
@@ -135,7 +142,9 @@ describe('ActorQueryOperationDescribeSubject', () => {
       };
       return actor.run(op).then(async(output: IQueryOperationResultQuads) => {
         expect(await output.metadata())
-          .toEqual({ cardinality: { type: 'estimate', value: 4 }, canContainUndefs: false });
+          .toEqual({ state: expect.any(MetadataValidationState),
+            cardinality: { type: 'estimate', value: 4 },
+            canContainUndefs: false });
         expect(output.type).toEqual('quads');
         expect(await arrayifyStream(output.quadStream)).toEqual([
           DF.quad(DF.namedNode('c'), DF.namedNode('__predicate'), DF.namedNode('__object')),

packages/actor-query-operation-nop/lib/ActorQueryOperationNop.ts

@@ -2,6 +2,7 @@ import { BindingsFactory } from '@comunica/bindings-factory';
 import type { IActorQueryOperationTypedMediatedArgs } from '@comunica/bus-query-operation';
 import { ActorQueryOperationTypedMediated } from '@comunica/bus-query-operation';
 import type { IActorTest } from '@comunica/core';
+import { MetadataValidationState } from '@comunica/metadata';
 import type { IActionContext, IQueryOperationResult } from '@comunica/types';
 import { SingletonIterator } from 'asynciterator';
 import type { Algebra } from 'sparqlalgebrajs';
@@ -24,6 +25,7 @@ export class ActorQueryOperationNop extends ActorQueryOperationTypedMediated<Alg
     return {
       bindingsStream: new SingletonIterator(BF.bindings()),
       metadata: () => Promise.resolve({
+        state: new MetadataValidationState(),
         cardinality: { type: 'exact', value: 1 },
         canContainUndefs: false,
         variables: [],

packages/actor-query-operation-nop/package.json

@@ -35,6 +35,7 @@
     "@comunica/bindings-factory": "^2.5.1",
     "@comunica/bus-query-operation": "^2.6.8",
     "@comunica/core": "^2.6.8",
+    "@comunica/metadata": "^2.5.0",
     "@comunica/types": "^2.6.8",
     "asynciterator": "^3.8.0",
     "sparqlalgebrajs": "^4.0.5"

packages/actor-query-operation-path-alt/test/ActorQueryOperationPathAlt-test.ts

@@ -1,6 +1,7 @@
 import { BindingsFactory } from '@comunica/bindings-factory';
 import { ActorQueryOperation } from '@comunica/bus-query-operation';
 import { ActionContext, Bus } from '@comunica/core';
+import { MetadataValidationState } from '@comunica/metadata';
 import { ArrayIterator } from 'asynciterator';
 import { DataFactory } from 'rdf-data-factory';
 import { Algebra, Factory } from 'sparqlalgebrajs';
@@ -25,6 +26,7 @@ describe('ActorQueryOperationPathAlt', () => {
           BF.bindings([[ DF.variable('x'), DF.literal('3') ]]),
         ]),
         metadata: () => Promise.resolve({
+          state: new MetadataValidationState(),
           cardinality: { type: 'estimate', value: 3 },
           canContainUndefs: false,
           variables: [ DF.variable('a') ],
@@ -88,6 +90,7 @@ describe('ActorQueryOperationPathAlt', () => {
       context: new ActionContext() };
       const output = ActorQueryOperation.getSafeBindings(await actor.run(op));
       expect(await output.metadata()).toEqual({
+        state: expect.any(MetadataValidationState),
         cardinality: { type: 'estimate', value: 6 },
         canContainUndefs: false,
         variables: [ DF.variable('a') ],

packages/actor-query-operation-path-zero-or-one/lib/ActorQueryOperationPathZeroOrOne.ts

@@ -2,6 +2,7 @@ import { ActorAbstractPath } from '@comunica/actor-abstract-path';
 import { BindingsFactory } from '@comunica/bindings-factory';
 import type { IActorQueryOperationTypedMediatedArgs } from '@comunica/bus-query-operation';
 import { ActorQueryOperation } from '@comunica/bus-query-operation';
+import { MetadataValidationState } from '@comunica/metadata';
 import type { Bindings, IQueryOperationResult, IActionContext } from '@comunica/types';
 import { SingletonIterator } from 'asynciterator';
 import { Algebra } from 'sparqlalgebrajs';
@@ -33,6 +34,7 @@ export class ActorQueryOperationPathZeroOrOne extends ActorAbstractPath {
         type: 'bindings',
         bindingsStream: new SingletonIterator(BF.bindings()),
         metadata: () => Promise.resolve({
+          state: new MetadataValidationState(),
           cardinality: { type: 'exact', value: 1 },
           canContainUndefs: false,
           variables: [],

packages/actor-query-operation-path-zero-or-one/package.json

@@ -35,6 +35,7 @@
     "@comunica/actor-abstract-path": "^2.6.8",
     "@comunica/bindings-factory": "^2.5.1",
     "@comunica/bus-query-operation": "^2.6.8",
+    "@comunica/metadata": "^2.5.0",
     "@comunica/types": "^2.6.8",
     "asynciterator": "^3.8.0",
     "sparqlalgebrajs": "^4.0.5"

packages/actor-query-operation-path-zero-or-one/test/ActorQueryOperationPathZeroOrOne-test.ts

@@ -2,6 +2,7 @@ import { BindingsFactory } from '@comunica/bindings-factory';
 import { ActorQueryOperation } from '@comunica/bus-query-operation';
 import { KeysQueryOperation } from '@comunica/context-entries';
 import { Bus, ActionContext } from '@comunica/core';
+import { MetadataValidationState } from '@comunica/metadata';
 import type * as RDF from '@rdfjs/types';
 import { ArrayIterator } from 'asynciterator';
 import { DataFactory } from 'rdf-data-factory';
@@ -51,6 +52,7 @@ describe('ActorQueryOperationPathZeroOrOne', () => {
         return Promise.resolve({
           bindingsStream: new ArrayIterator(distinct ? [ bindings[0] ] : bindings),
           metadata: () => Promise.resolve({
+            state: new MetadataValidationState(),
             cardinality: { type: 'estimate', value: distinct ? 1 : 3 },
             canContainUndefs: false,
             variables: vars,
@@ -107,6 +109,7 @@ describe('ActorQueryOperationPathZeroOrOne', () => {
       context: new ActionContext() };
       const output = ActorQueryOperation.getSafeBindings(await actor.run(op));
       expect(await output.metadata()).toEqual({
+        state: expect.any(MetadataValidationState),
         cardinality: { type: 'estimate', value: 1 },
         canContainUndefs: false,
         variables: [ DF.variable('x') ],
@@ -125,6 +128,7 @@ describe('ActorQueryOperationPathZeroOrOne', () => {
       context: new ActionContext({ [KeysQueryOperation.isPathArbitraryLengthDistinctKey.name]: false }) };
       const output = ActorQueryOperation.getSafeBindings(await actor.run(op));
       expect(await output.metadata()).toEqual({
+        state: expect.any(MetadataValidationState),
         cardinality: { type: 'estimate', value: 1 },
         canContainUndefs: false,
         variables: [ DF.variable('x') ],
@@ -143,6 +147,7 @@ describe('ActorQueryOperationPathZeroOrOne', () => {
       context: new ActionContext({ [KeysQueryOperation.isPathArbitraryLengthDistinctKey.name]: true }) };
       const output = ActorQueryOperation.getSafeBindings(await actor.run(op));
       expect(await output.metadata()).toEqual({
+        state: expect.any(MetadataValidationState),
         cardinality: { type: 'estimate', value: 3 },
         canContainUndefs: false,
         variables: [ DF.variable('x') ],
@@ -164,6 +169,7 @@ describe('ActorQueryOperationPathZeroOrOne', () => {
       context: new ActionContext({ [KeysQueryOperation.isPathArbitraryLengthDistinctKey.name]: true }) };
       const output = ActorQueryOperation.getSafeBindings(await actor.run(op));
       expect(await output.metadata()).toEqual({
+        state: expect.any(MetadataValidationState),
         cardinality: { type: 'estimate', value: 3 },
         canContainUndefs: false,
         variables: [ DF.variable('x') ],
@@ -185,6 +191,7 @@ describe('ActorQueryOperationPathZeroOrOne', () => {
       context: new ActionContext({ [KeysQueryOperation.isPathArbitraryLengthDistinctKey.name]: true }) };
       const output = ActorQueryOperation.getSafeBindings(await actor.run(op));
       expect(await output.metadata()).toEqual({
+        state: expect.any(MetadataValidationState),
         cardinality: { type: 'estimate', value: 3 },
         canContainUndefs: false,
         variables: [],
@@ -203,6 +210,7 @@ describe('ActorQueryOperationPathZeroOrOne', () => {
       context: new ActionContext({ [KeysQueryOperation.isPathArbitraryLengthDistinctKey.name]: true }) };
       const output = ActorQueryOperation.getSafeBindings(await actor.run(op));
       expect(await output.metadata()).toEqual({
+        state: expect.any(MetadataValidationState),
         cardinality: { type: 'exact', value: 1 },
         canContainUndefs: false,
         variables: [],

packages/actor-query-operation-service/lib/ActorQueryOperationService.ts

@@ -3,6 +3,7 @@ import type { IActorQueryOperationTypedMediatedArgs } from '@comunica/bus-query-
 import { ActorQueryOperation, ActorQueryOperationTypedMediated } from '@comunica/bus-query-operation';
 import { KeysInitQuery, KeysRdfResolveQuadPattern } from '@comunica/context-entries';
 import type { IActorTest } from '@comunica/core';
+import { MetadataValidationState } from '@comunica/metadata';
 import type { IActionContext, IQueryOperationResult, IQueryOperationResultBindings } from '@comunica/types';
 import { SingletonIterator } from 'asynciterator';
 import type { Algebra } from 'sparqlalgebrajs';
@@ -50,7 +51,12 @@ export class ActorQueryOperationService extends ActorQueryOperationTypedMediated
         output = {
           bindingsStream: new SingletonIterator(BF.bindings()),
           type: 'bindings',
-          metadata: async() => ({ cardinality: { type: 'exact', value: 1 }, canContainUndefs: false, variables: []}),
+          metadata: async() => ({
+            state: new MetadataValidationState(),
+            cardinality: { type: 'exact', value: 1 },
+            canContainUndefs: false,
+            variables: [],
+          }),
         };
       } else {
         throw error;