busify orderBy & add extensionFunction & move aggregateEval & functio…

…n in single bus

packages/actor-bindings-aggregator-factory-average/lib/ActorBindingsAggregatorFactoryAverage.ts

@@ -5,6 +5,7 @@ import type {
 import { ActorBindingsAggregatorFactory } from '@comunica/bus-bindings-aggeregator-factory';
 import type { IActorTest } from '@comunica/core';
 import { RegularOperator } from '@comunica/expression-evaluator';
+import type { ITermFunction } from '@comunica/types';
 import { AverageAggregator } from './AverageAggregator';
 
 /**
@@ -22,13 +23,22 @@ export class ActorBindingsAggregatorFactoryAverage extends ActorBindingsAggregat
     return {};
   }
 
-  public async run(action: IActionBindingsAggregatorFactory): Promise<IActorBindingsAggregatorFactoryOutput> {
+  public async run({ factory, context, expr }: IActionBindingsAggregatorFactory):
+  Promise<IActorBindingsAggregatorFactoryOutput> {
     return {
       aggregator: new AverageAggregator(
-        await action.factory.createEvaluator(action.expr, action.context),
-        action.expr.distinct,
-        await action.factory.createTermFunction({ functionName: RegularOperator.ADDITION }),
-        await action.factory.createTermFunction({ functionName: RegularOperator.DIVISION }),
+        await factory.createEvaluator(expr, context),
+        expr.distinct,
+        <ITermFunction> await factory.createFunction({
+          functionName: RegularOperator.ADDITION,
+          context,
+          definitionType: 'onTerm',
+        }),
+        <ITermFunction> await factory.createFunction({
+          functionName: RegularOperator.DIVISION,
+          context,
+          definitionType: 'onTerm',
+        }),
       ),
     };
   }

packages/actor-bindings-aggregator-factory-average/lib/AverageAggregator.ts

@@ -1,8 +1,7 @@
 import type { ExpressionEvaluator } from '@comunica/expression-evaluator';
 import { AggregateEvaluator, typedLiteral, TypeURL } from '@comunica/expression-evaluator';
 import * as E from '@comunica/expression-evaluator/lib/expressions';
-import type { RegularFunction } from '@comunica/expression-evaluator/lib/functions';
-import type { IBindingsAggregator, IExpressionEvaluator } from '@comunica/types';
+import type { IBindingsAggregator, IExpressionEvaluator, ITermFunction } from '@comunica/types';
 import type * as RDF from '@rdfjs/types';
 
 interface IAverageState {
@@ -16,8 +15,8 @@ export class AverageAggregator extends AggregateEvaluator implements IBindingsAg
   public constructor(
     evaluator: IExpressionEvaluator,
     distinct: boolean,
-    private readonly additionFunction: RegularFunction,
-    private readonly divisionFunction: RegularFunction,
+    private readonly additionFunction: ITermFunction,
+    private readonly divisionFunction: ITermFunction,
     throwError?: boolean,
   ) {
     super(evaluator, distinct, throwError);

packages/actor-bindings-aggregator-factory-sum/lib/ActorBindingsAggregatorFactorySum.ts

@@ -8,6 +8,7 @@ import {
 } from '@comunica/bus-bindings-aggeregator-factory';
 import type { IActorTest } from '@comunica/core';
 import { RegularOperator } from '@comunica/expression-evaluator';
+import type { ITermFunction } from '@comunica/types';
 import { SumAggregator } from './SumAggregator';
 
 /**
@@ -25,12 +26,17 @@ export class ActorBindingsAggregatorFactorySum extends ActorBindingsAggregatorFa
     return {};
   }
 
-  public async run(action: IActionBindingsAggregatorFactory): Promise<IActorBindingsAggregatorFactoryOutput> {
+  public async run({ factory, expr, context }: IActionBindingsAggregatorFactory):
+  Promise<IActorBindingsAggregatorFactoryOutput> {
     return {
       aggregator: new SumAggregator(
-        await action.factory.createEvaluator(action.expr, action.context),
-        action.expr.distinct,
-        await action.factory.createTermFunction({ functionName: RegularOperator.ADDITION }),
+        await factory.createEvaluator(expr, context),
+        expr.distinct,
+        <ITermFunction> await factory.createFunction({
+          functionName: RegularOperator.ADDITION,
+          context,
+          definitionType: 'onTerm',
+        }),
       ),
     };
   }

packages/actor-bindings-aggregator-factory-sum/lib/SumAggregator.ts

@@ -1,8 +1,7 @@
 import type { ExpressionEvaluator } from '@comunica/expression-evaluator';
 import { AggregateEvaluator, typedLiteral, TypeURL } from '@comunica/expression-evaluator';
 import type * as E from '@comunica/expression-evaluator/lib/expressions';
-import type { RegularFunction } from '@comunica/expression-evaluator/lib/functions';
-import type { IExpressionEvaluator } from '@comunica/types';
+import type { IExpressionEvaluator, ITermFunction } from '@comunica/types';
 import type * as RDF from '@rdfjs/types';
 
 type SumState = E.NumericLiteral;
@@ -12,7 +11,7 @@ export class SumAggregator extends AggregateEvaluator {
 
   public constructor(evaluator: IExpressionEvaluator,
     distinct: boolean,
-    private readonly additionFunction: RegularFunction,
+    private readonly additionFunction: ITermFunction,
     throwError?: boolean) {
     super(evaluator, distinct, throwError);
   }

packages/expression-evaluator/lib/evaluators/AggregateEvaluator.ts → packages/bus-bindings-aggregator-factory/lib/AggregateEvaluator.ts

@@ -1,12 +1,12 @@
+import type { ExpressionEvaluator } from '@comunica/expression-evaluator/lib/evaluators/ExpressionEvaluator';
+import type * as E from '@comunica/expression-evaluator/lib/expressions';
+import { TermTransformer } from '@comunica/expression-evaluator/lib/transformers/TermTransformer';
+import { TypeAlias } from '@comunica/expression-evaluator/lib/util/Consts';
+import { EmptyAggregateError } from '@comunica/expression-evaluator/lib/util/Errors';
+import { isSubTypeOf } from '@comunica/expression-evaluator/lib/util/TypeHandling';
 import type { IExpressionEvaluator } from '@comunica/types';
 import type * as RDF from '@rdfjs/types';
 import * as RdfString from 'rdf-string';
-import type * as E from '../expressions';
-import { TermTransformer } from '../transformers/TermTransformer';
-import { TypeAlias } from '../util/Consts';
-import { EmptyAggregateError } from '../util/Errors';
-import { isSubTypeOf } from '../util/TypeHandling';
-import type { ExpressionEvaluator } from './ExpressionEvaluator';
 
 /**
  * This is the base class for all aggregators.

packages/context-entries/lib/Keys.ts

@@ -137,7 +137,7 @@ export const KeysInitQuery = {
    * The dictionary-based extensionFunctions context entry may be used instead, but not simultaneously.
    */
   extensionFunctionCreator: new ActionContextKey<
-  (functionNamedNode: RDF.NamedNode) => ((args: RDF.Term[]) => Promise<RDF.Term>) | undefined
+  (functionNamedNode: RDF.NamedNode) => Promise<((args: RDF.Term[]) => Promise<RDF.Term>) | undefined>
   // eslint-disable-next-line @typescript-eslint/no-extra-parens
   >('@comunica/actor-init-query:extensionFunctionCreator'),
   /**

packages/expression-evaluator/lib/evaluators/ContextualizedEvaluator.ts

@@ -1,9 +1,8 @@
 import type { MediatorQueryOperation } from '@comunica/bus-query-operation';
 import { ActorQueryOperation, materializeOperation } from '@comunica/bus-query-operation';
-import type { FunctionBusType, IActionContext, TermFunctionBusType } from '@comunica/types';
+import type { FunctionBusType, IActionContext } from '@comunica/types';
 import type * as RDF from '@rdfjs/types';
 import { LRUCache } from 'lru-cache';
-import type { Algebra as Alg } from 'sparqlalgebrajs';
 import * as E from '../expressions';
 import { expressionToVar } from '../functions/Helpers';
 import type { FunctionArgumentsCache } from '../functions/OverloadTree';
@@ -14,10 +13,10 @@ import * as Err from '../util/Errors';
 import type { SuperTypeCallback, TypeCache, ISuperTypeProvider } from '../util/TypeHandling';
 
 export type AsyncExtensionFunction = (args: RDF.Term[]) => Promise<RDF.Term>;
-export type AsyncExtensionFunctionCreator = (functionNamedNode: RDF.NamedNode) => AsyncExtensionFunction | undefined;
+export type AsyncExtensionFunctionCreator = (functionNamedNode: RDF.NamedNode) =>
+Promise<AsyncExtensionFunction | undefined>;
 
 export interface IAsyncEvaluatorContext {
-  extensionFunctionCreator?: AsyncExtensionFunctionCreator;
   now?: Date;
   baseIRI?: string;
   typeCache?: TypeCache;
@@ -26,7 +25,6 @@ export interface IAsyncEvaluatorContext {
   defaultTimeZone?: ITimeZoneRepresentation;
   actionContext: IActionContext;
   mediatorQueryOperation: MediatorQueryOperation;
-  mediatorTermFunction: TermFunctionBusType;
   mediatorFunction: FunctionBusType;
 }
 
@@ -35,7 +33,7 @@ export interface IAsyncEvaluatorContext {
  * It also holds all context items needed for evaluating functions.
  */
 export class ContextualizedEvaluator {
-  protected readonly transformer: AlgebraTransformer;
+  public readonly transformer: AlgebraTransformer;
 
   private readonly subEvaluators: Record<string,
   (expr: E.Expression, mapping: RDF.Bindings) => Promise<E.Term> | E.Term> =
@@ -47,7 +45,6 @@ export class ContextualizedEvaluator {
         [E.ExpressionType.Named]: this.evalFunction.bind(this),
         [E.ExpressionType.Existence]: this.evalExistence.bind(this),
         [E.ExpressionType.Aggregate]: this.evalAggregate.bind(this),
-        [E.ExpressionType.AsyncExtension]: this.evalFunction.bind(this),
       };
 
   // Context items
@@ -59,7 +56,6 @@ export class ContextualizedEvaluator {
   public readonly defaultTimeZone: ITimeZoneRepresentation;
   public readonly actionContext: IActionContext;
   public readonly mediatorQueryOperation: MediatorQueryOperation;
-  public readonly mediatorTermFunction: TermFunctionBusType;
   public readonly mediatorFunction: FunctionBusType;
 
   public constructor(context: IAsyncEvaluatorContext) {
@@ -70,24 +66,17 @@ export class ContextualizedEvaluator {
       cache: context.typeCache || new LRUCache({ max: 1_000 }),
       discoverer: context.getSuperType || (() => 'term'),
     };
-    // eslint-disable-next-line unicorn/no-useless-undefined
-    this.extensionFunctionCreator = context.extensionFunctionCreator || (() => undefined);
     this.defaultTimeZone = context.defaultTimeZone || extractTimeZone(this.now);
     this.actionContext = context.actionContext;
     this.mediatorQueryOperation = context.mediatorQueryOperation;
-    this.mediatorTermFunction = context.mediatorTermFunction;
     this.mediatorFunction = context.mediatorFunction;
 
     this.transformer = new AlgebraTransformer(
       this.superTypeProvider,
-      this.mediatorFunction,
+      args => this.mediatorFunction({ ...args, context: this.actionContext }),
     );
   }
 
-  public translate(algExpr: Alg.Expression): Promise<E.Expression> {
-    return this.transformer.transformAlgebra(algExpr);
-  }
-
   public async evaluateAsInternal(expr: E.Expression, mapping: RDF.Bindings): Promise<E.Term> {
     const evaluator = this.subEvaluators[expr.expressionType];
     if (!evaluator) {
@@ -108,7 +97,7 @@ export class ContextualizedEvaluator {
     return this.transformer.transformRDFTermUnsafe(term);
   }
 
-  private async evalFunction(expr: E.Operator | E.SpecialOperator | E.Named | E.AsyncExtension, mapping: RDF.Bindings):
+  private async evalFunction(expr: E.Operator | E.SpecialOperator | E.Named, mapping: RDF.Bindings):
   Promise<E.Term> {
     return expr.apply({
       args: expr.args,

packages/expression-evaluator/lib/evaluators/ExpressionEvaluatorFactory.ts

@@ -1,50 +1,69 @@
 import type { MediatorBindingsAggregatorFactory } from '@comunica/bus-bindings-aggeregator-factory';
 import type { MediatorQueryOperation } from '@comunica/bus-query-operation';
 import { KeysInitQuery } from '@comunica/context-entries';
-import type {
-  IActionContext,
+import type { IActionContext,
   IBindingsAggregator,
   IExpressionEvaluator,
   IExpressionEvaluatorFactory,
-  FunctionBusType, TermFunctionBusType, IOrderByEvaluator,
-} from '@comunica/types';
+  FunctionBusType, IOrderByEvaluator,
+  FunctionExpression, ITermFunction, OrderByBus,
+  IEvalContext } from '@comunica/types';
+import type * as RDF from '@rdfjs/types';
+import { DataFactory } from 'rdf-data-factory';
 import type { Algebra as Alg } from 'sparqlalgebrajs';
 import type * as E from '../expressions';
-import type { TermSparqlFunction, SparqlFunction } from '../functions';
-import { namedFunctions, regularFunctions, specialFunctions } from '../functions';
+import { FunctionDefinition, namedFunctions, regularFunctions, specialFunctions } from '../functions';
 import type * as C from '../util/Consts';
 import { RegularOperator } from '../util/Consts';
-import type { IAsyncEvaluatorContext } from './ContextualizedEvaluator';
+import type { IAsyncEvaluatorContext, AsyncExtensionFunction } from './ContextualizedEvaluator';
 import { ContextualizedEvaluator } from './ContextualizedEvaluator';
 import { ExpressionEvaluator } from './ExpressionEvaluator';
 import { OrderByEvaluator } from './OrderByEvaluator';
 
 export class ExpressionEvaluatorFactory implements IExpressionEvaluatorFactory {
   public readonly mediatorBindingsAggregatorFactory: MediatorBindingsAggregatorFactory;
   public readonly mediatorQueryOperation: MediatorQueryOperation;
-  public readonly functionsBus: FunctionBusType = async({ functionName }) => {
-    const res: SparqlFunction | undefined = {
+  public readonly functionsBus: FunctionBusType = async({ functionName, context }) => {
+    const res: FunctionExpression | undefined = {
       ...regularFunctions,
       ...specialFunctions,
       ...namedFunctions,
     }[<C.NamedOperator | C.Operator> functionName];
     if (res) {
       return res;
     }
-    throw new Error('nah!');
-  };
-
-  public readonly termFunctionsBus: TermFunctionBusType = async({ functionName }) => {
-    const res: TermSparqlFunction<any> | undefined = {
-      ...regularFunctions,
-      ...namedFunctions,
-    }[<C.NamedOperator | C.RegularOperator> functionName];
-    if (res) {
-      return res;
+    const extensionFinder: ((functionNamedNode: RDF.NamedNode) =>
+    Promise<((args: RDF.Term[]) => Promise<RDF.Term>) | undefined>) | undefined =
+      context.get(KeysInitQuery.extensionFunctionCreator);
+    if (extensionFinder) {
+      const definition = await extensionFinder(new DataFactory<RDF.Quad>().namedNode(functionName));
+      if (definition) {
+        return new NamedExtension(definition);
+      }
+    }
+    const extensionMap: Record<string, (args: RDF.Term[]) => Promise<RDF.Term>> | undefined =
+      context.get(KeysInitQuery.extensionFunctions);
+    if (extensionMap) {
+      const definition = extensionMap[functionName];
+      if (definition) {
+        return new NamedExtension(definition);
+      }
     }
     throw new Error('nah!');
   };
 
+  public readonly orderByBus: OrderByBus = async({ context }) =>
+    new OrderByEvaluator(new ContextualizedEvaluator({
+      now: context.get(KeysInitQuery.queryTimestamp),
+      baseIRI: context.get(KeysInitQuery.baseIRI),
+      functionArgumentsCache: context.get(KeysInitQuery.functionArgumentsCache),
+      actionContext: context,
+      mediatorQueryOperation: this.mediatorQueryOperation,
+      mediatorFunction: this.functionsBus,
+    }),
+    <ITermFunction> await this.functionsBus({ functionName: RegularOperator.EQUAL, context, definitionType: 'onTerm' }),
+    <ITermFunction> await this.functionsBus({ functionName: RegularOperator.LT, context, definitionType: 'onTerm' }));
+
   public constructor(args: IExpressionEvaluatorFactoryArgs) {
     this.mediatorBindingsAggregatorFactory = args.mediatorBindingsAggregatorFactory;
     this.mediatorQueryOperation = args.mediatorQueryOperation;
@@ -62,12 +81,11 @@ export class ExpressionEvaluatorFactory implements IExpressionEvaluatorFactory {
         actionContext: context,
         mediatorQueryOperation: this.mediatorQueryOperation,
         mediatorFunction: this.functionsBus,
-        mediatorTermFunction: this.termFunctionsBus,
         ...legacyContext,
       });
-      return new ExpressionEvaluator(defContextEval, await defContextEval.translate(algExpr));
+      return new ExpressionEvaluator(defContextEval, await defContextEval.transformer.transformAlgebra(algExpr));
     }
-    return new ExpressionEvaluator(contextEval, await contextEval.translate(algExpr));
+    return new ExpressionEvaluator(contextEval, await contextEval.transformer.transformAlgebra(algExpr));
   }
 
   public async createAggregator(algExpr: Alg.AggregateExpression, context: IActionContext):
@@ -79,33 +97,36 @@ export class ExpressionEvaluatorFactory implements IExpressionEvaluatorFactory {
     })).aggregator;
   }
 
-  public createTermFunction(arg: { functionName: string; arguments?: E.TermExpression[] }):
-  Promise<TermSparqlFunction<any>> {
-    return this.termFunctionsBus(arg);
-  }
-
-  public createFunction(arg: { functionName: string; arguments: Alg.Expression[] }): Promise<SparqlFunction> {
-    return this.functionsBus(arg);
-  }
+  public createFunction = this.functionsBus;
 
-  public async createOrderByEvaluator(context: IActionContext, legacyContext: Partial<IAsyncEvaluatorContext> = {}):
+  public async createOrderByEvaluator(context: IActionContext):
   Promise<IOrderByEvaluator> {
-    return new OrderByEvaluator({
-      now: context.get(KeysInitQuery.queryTimestamp),
-      baseIRI: context.get(KeysInitQuery.baseIRI),
-      functionArgumentsCache: context.get(KeysInitQuery.functionArgumentsCache),
-      actionContext: context,
-      mediatorQueryOperation: this.mediatorQueryOperation,
-      mediatorFunction: this.functionsBus,
-      mediatorTermFunction: this.termFunctionsBus,
-      ...legacyContext,
-    },
-    await this.createTermFunction({ functionName: RegularOperator.EQUAL }),
-    await this.createTermFunction({ functionName: RegularOperator.LT }));
+    return this.orderByBus({ context });
   }
 }
 
 interface IExpressionEvaluatorFactoryArgs {
   mediatorBindingsAggregatorFactory: MediatorBindingsAggregatorFactory;
   mediatorQueryOperation: MediatorQueryOperation;
 }
+
+// TODO: this thing will be it's own actor but it's just a little special.
+//  It will also be the only consumer of the context items:
+//  KeysInitQuery.extensionFunctions and KeysInitQuery.extensionFunctionCreator
+class NamedExtension extends FunctionDefinition {
+  // TODO: the context should be checked in the test part of the actor.
+  //  The fact that this can be done is async now is a nice feature!
+  //  It means that named function definitions could be queried over the web!
+  // TODO: when all is done, this should be injected in some way!
+  protected arity = Number.POSITIVE_INFINITY;
+  public constructor(private readonly functionDefinition: AsyncExtensionFunction) {
+    super();
+  }
+
+  public apply = async({ args, exprEval, mapping }: IEvalContext): Promise<E.TermExpression> => {
+    const evaluatedArgs: E.Term[] = await Promise.all(args.map(arg => exprEval.evaluateAsInternal(arg, mapping)));
+    return exprEval.transformer.transformRDFTermUnsafe(
+      await this.functionDefinition(evaluatedArgs.map(term => term.toRDF())),
+    );
+  };
+}