Add support for parallel execution of readonly functions (#2)

Author: daniel-chambers

CHANGELOG.md

@@ -4,6 +4,8 @@ This changelog documents the changes between release versions.
 ## main
 Changes to be included in the next upcoming release
 
+- Add support for parallel execution of readonly functions ([#2](https://github.com/hasura/ndc-nodejs-lambda/pull/2))
+
 ## v0.10.0
 - Add missing query.variables capability
 

README.md

@@ -19,6 +19,17 @@ The `package.config` has been created with `start` and `watch` scripts. These us
 ### Functions
 Any functions exported from `functions.ts` are made available as NDC functions/procedures to use in your Hasura metadata and expose as GraphQL fields in queries or mutation.
 
+If you write a function that performs a read-only operation, you should mark it with the `@readonly` JSDoc tag, and it will be exposed as an NDC function, which will ultimately show up as a GraphQL query field in Hasura.
+
+```typescript
+/** @readonly */
+export function add(x: number, y: number): number {
+  return x + y;
+}
+```
+
+Functions without the `@readonly` JSDoc tag are exposed as NDC procedures, which will ultimately show up as a GraphQL mutation field in Hasura.
+
 Arguments to the function end up being field arguments in GraphQL and the return value is what the field will return when queried. Every function must return a value; `void`, `null` or `undefined` is not supported.
 
 ```typescript
@@ -136,18 +147,24 @@ These types are unsupported as function parameter types or return types for func
 * [`void`](https://www.typescriptlang.org/docs/handbook/2/functions.html#void), [`object`](https://www.typescriptlang.org/docs/handbook/2/functions.html#object), [`unknown`](https://www.typescriptlang.org/docs/handbook/2/functions.html#unknown), [`never`](https://www.typescriptlang.org/docs/handbook/2/functions.html#never), [`any`](https://www.typescriptlang.org/docs/handbook/2/everyday-types.html#any) types - to accept and return arbitrary JSON, use `sdk.JSONValue` instead
 * `null` and `undefined` - unless used in a union with a single other type
 
-
-### Impure/pure functions
-If you write a function that performs a read-only operation, you should mark it with the `@readonly` JSDoc tag, and it will be exposed as an NDC function, which will ultimately show up as a GraphQL query field in Hasura.
-
-```typescript
-/** @readonly */
-export function add(x: number, y: number): number {
-  return x + y;
+### Parallel execution
+If functions are involved remote relationships in your Hasura metadata, then they may be queried in a [batch-based fashion](https://hasura.github.io/ndc-spec/specification/queries/variables.html). In this situation, any async functions that are marked with the `@readonly` JSDoc tag may be executed in parallel. The default degree of parallelism per query request to the connector is 10, but you may customise this by using the `@paralleldegree` JSDoc tag on your function.
+
+``` typescript
+/**
+ * This function will only run up to 5 http requests in parallel per query
+ *
+ * @readonly
+ * @paralleldegree 5
+ */
+export async function test(statusCode: number): Promise<string> {
+  const result = await fetch("http://httpstat.us/${statusCode}")
+  const responseBody = await result.json() as any;
+  return responseBody.description;
 }
 ```
 
-Functions without the `@readonly` JSDoc tag are exposed as NDC procedures, which will ultimately show up as a GraphQL mutation field in Hasura.
+Non-readonly functions are not invoked in parallel within the same mutation request to the connector, so it is invalid to use the @paralleldegree JSDoc tag on those functions.
 
 ## Deploying with `hasura3 connector create`
 

ndc-lambda-sdk/package-lock.json

@@ -35,6 +35,7 @@
     "@tsconfig/node18": "^18.2.2",
     "commander": "^11.1.0",
     "cross-spawn": "^7.0.3",
+    "p-limit": "^3.1.0",
     "ts-api-utils": "^1.0.3",
     "ts-node": "^10.9.2",
     "ts-node-dev": "^2.0.0",

ndc-lambda-sdk/package.json

@@ -1,11 +1,16 @@
 import * as sdk from "@hasura/ndc-sdk-typescript"
+import pLimit from "p-limit";
 import * as schema from "./schema"
 import { isArray, mapObjectValues, unreachable } from "./util"
 
 export type RuntimeFunctions = {
   [functionName: string]: Function
 }
 
+// This number is chosen arbitrarily, just to place _some_ limit on the amount of
+// parallelism going on within a single query
+const DEFAULT_PARALLEL_DEGREE = 10;
+
 export async function executeQuery(queryRequest: sdk.QueryRequest, functionsSchema: schema.FunctionsSchema, runtimeFunctions: RuntimeFunctions): Promise<sdk.QueryResponse> {
   const functionName = queryRequest.collection;
 
@@ -25,21 +30,21 @@ export async function executeQuery(queryRequest: sdk.QueryRequest, functionsSche
     return prepareArguments(resolvedArgs, functionDefinition, functionsSchema.objectTypes);
   });
 
-  const rowSets: sdk.RowSet[] = [];
-  for (const invocationPreparedArgs of functionInvocationPreparedArgs) {
+  const parallelLimit = pLimit(functionDefinition.parallelDegree ?? DEFAULT_PARALLEL_DEGREE);
+  const functionInvocations: Promise<sdk.RowSet>[] = functionInvocationPreparedArgs.map(invocationPreparedArgs => parallelLimit(async () => {
     const result = await invokeFunction(runtimeFunction, invocationPreparedArgs, functionName);
     const prunedResult = reshapeResultToNdcResponseValue(result, functionDefinition.resultType, [], queryRequest.query.fields ?? {}, functionsSchema.objectTypes);
-    rowSets.push({
+    return {
       aggregates: {},
       rows: [
         {
           __value: prunedResult
         }
       ]
-    });
-  }
+    };
+  }));
 
-  return rowSets;
+  return await Promise.all(functionInvocations);
 }
 
 export async function executeMutation(mutationRequest: sdk.MutationRequest, functionsSchema: schema.FunctionsSchema, runtimeFunctions: RuntimeFunctions): Promise<sdk.MutationResponse> {

ndc-lambda-sdk/src/execution.ts

@@ -168,6 +168,7 @@ function deriveFunctionSchema(functionDeclaration: ts.FunctionDeclaration, expor
 
   const functionDescription = ts.displayPartsToString(functionSymbol.getDocumentationComment(context.typeChecker)).trim();
   const markedReadonlyInJsDoc = functionSymbol.getJsDocTags().find(e => e.name === "readonly") !== undefined;
+  const parallelDegreeResult = getParallelDegreeFromJsDoc(functionSymbol, markedReadonlyInJsDoc);
 
   const functionCallSig = functionType.getCallSignatures()[0] ?? throwError(`Function '${exportedFunctionName}' didn't have a call signature`)
   const functionSchemaArguments: Result<schema.ArgumentDefinition[], string[]> = Result.traverseAndCollectErrors(functionCallSig.getParameters(), paramSymbol => {
@@ -187,15 +188,37 @@ function deriveFunctionSchema(functionDeclaration: ts.FunctionDeclaration, expor
   const returnType = functionCallSig.getReturnType();
   const returnTypeResult = deriveSchemaTypeForTsType(unwrapPromiseType(returnType, context.typeChecker) ?? returnType, [{segmentType: "FunctionReturn", functionName: exportedFunctionName}], context);
 
-  return Result.collectErrors(functionSchemaArguments, returnTypeResult)
-    .map(([functionSchemaArgs, returnType]) => ({
+  return Result.collectErrors3(functionSchemaArguments, returnTypeResult, parallelDegreeResult)
+    .map(([functionSchemaArgs, returnType, parallelDegree]) => ({
       description: functionDescription ? functionDescription : null,
       ndcKind: markedReadonlyInJsDoc ? schema.FunctionNdcKind.Function : schema.FunctionNdcKind.Procedure,
       arguments: functionSchemaArgs,
-      resultType: returnType
+      resultType: returnType,
+      parallelDegree,
     }));
 }
 
+function getParallelDegreeFromJsDoc(functionSymbol: ts.Symbol, functionIsReadonly: boolean): Result<number | null, string[]> {
+  const parallelDegreeTag = functionSymbol.getJsDocTags().find(e => e.name === "paralleldegree");
+  if (parallelDegreeTag === undefined) {
+    return new Ok(null);
+  } else {
+    if (!functionIsReadonly)
+      return new Err(["The @paralleldegree JSDoc tag is only supported on functions also marked with the @readonly JSDoc tag"]);
+
+    const tagSymbolDisplayPart = parallelDegreeTag.text?.[0]
+    if (tagSymbolDisplayPart === undefined)
+      return new Err(["The @paralleldegree JSDoc tag must specify an integer degree value"]);
+
+    const tagText = tagSymbolDisplayPart.text.trim();
+    const parallelDegreeInt = parseInt(tagText, 10);
+    if (isNaN(parallelDegreeInt) || parallelDegreeInt <= 0)
+      return new Err([`The @paralleldegree JSDoc tag must specify an integer degree value that is greater than 0. Current value: '${tagText}'`]);
+
+    return new Ok(parallelDegreeInt);
+  }
+}
+
 const MAX_TYPE_DERIVATION_RECURSION = 20; // Better to abort than get into an infinite loop, this could be increased if required.
 
 type TypePathSegment =

ndc-lambda-sdk/src/inference.ts

@@ -15,7 +15,8 @@ export type FunctionDefinition = {
   ndcKind: FunctionNdcKind
   description: string | null,
   arguments: ArgumentDefinition[] // Function arguments are ordered
-  resultType: TypeDefinition
+  resultType: TypeDefinition,
+  parallelDegree: number | null,
 }
 
 export enum FunctionNdcKind {

ndc-lambda-sdk/src/schema.ts

@@ -29,3 +29,7 @@ export function getFlags(flagsEnum: Record<string, string | number>, value: numb
           : []
     });
 }
+
+export function sleep(ms: number): Promise<void> {
+  return new Promise((r) => setTimeout(r, ms))
+}

ndc-lambda-sdk/src/util.ts

@@ -3,6 +3,7 @@ import { assert } from "chai";
 import * as sdk from "@hasura/ndc-sdk-typescript"
 import { executeQuery } from "../../src/execution";
 import { FunctionNdcKind, FunctionsSchema } from "../../src/schema";
+import { sleep } from "../../src/util";
 
 describe("execute query", function() {
   it("executes the function", async function() {
@@ -18,6 +19,7 @@ describe("execute query", function() {
         "theFunction": {
           ndcKind: FunctionNdcKind.Function,
           description: null,
+          parallelDegree: null,
           arguments: [
             {
               argumentName: "param",
@@ -80,6 +82,7 @@ describe("execute query", function() {
         "theFunction": {
           ndcKind: FunctionNdcKind.Function,
           description: null,
+          parallelDegree: null,
           arguments: [
             {
               argumentName: "param",
@@ -154,5 +157,118 @@ describe("execute query", function() {
       }
     ]);
     assert.equal(functionCallCount, 2);
-  })
+  });
+
+  it("when there are variables, executes the function in parallel, but respecting the configured parallel degree", async function() {
+    let functionCallCompletions: string[] = [];
+    const runtimeFunctions = {
+      "theFunction": async (invocationName: string, sleepMs: number) => {
+        await sleep(sleepMs);
+        functionCallCompletions.push(invocationName)
+        return invocationName;
+      }
+    };
+    const functionSchema: FunctionsSchema = {
+      functions: {
+        "theFunction": {
+          ndcKind: FunctionNdcKind.Function,
+          description: null,
+          parallelDegree: 3,
+          arguments: [
+            {
+              argumentName: "invocationName",
+              description: null,
+              type: {
+                type: "named",
+                kind: "scalar",
+                name: "String"
+              }
+            },
+            {
+              argumentName: "sleepMs",
+              description: null,
+              type: {
+                type: "named",
+                kind: "scalar",
+                name: "Float"
+              }
+            },
+          ],
+          resultType: {
+            type: "named",
+            kind: "scalar",
+            name: "String"
+          },
+        }
+      },
+      objectTypes: {},
+      scalarTypes: {
+        "String": {}
+      }
+    };
+    const queryRequest: sdk.QueryRequest = {
+      collection: "theFunction",
+      query: {
+        fields: {},
+      },
+      arguments: {
+        "invocationName": {
+          type: "variable",
+          name: "invocationName"
+        },
+        "sleepMs": {
+          type: "variable",
+          name: "sleepMs"
+        }
+      },
+      collection_relationships: {},
+      variables: [
+        {
+          "invocationName": "first",
+          "sleepMs": 100,
+        },
+        {
+          "invocationName": "second",
+          "sleepMs": 250,
+        },
+        {
+          "invocationName": "third",
+          "sleepMs": 150,
+        },
+        {
+          "invocationName": "fourth",
+          "sleepMs": 100,
+        },
+      ]
+    };
+
+    const result = await executeQuery(queryRequest, functionSchema, runtimeFunctions);
+    assert.deepStrictEqual(result, [
+      {
+        aggregates: {},
+        rows: [
+          { __value: "first" }
+        ]
+      },
+      {
+        aggregates: {},
+        rows: [
+          { __value: "second" }
+        ]
+      },
+      {
+        aggregates: {},
+        rows: [
+          { __value: "third" }
+        ]
+      },
+      {
+        aggregates: {},
+        rows: [
+          { __value: "fourth" }
+        ]
+      },
+    ]);
+    assert.deepStrictEqual(functionCallCompletions, ["first", "third", "fourth", "second"]);
+  });
 });

ndc-lambda-sdk/test/execution/execute-query.test.ts

@@ -9,6 +9,7 @@ describe("prepare arguments", function() {
     const functionDefinition: FunctionDefinition = {
       ndcKind: FunctionNdcKind.Function,
       description: null,
+      parallelDegree: null,
       arguments: [
         {
           argumentName: "c",
@@ -60,6 +61,7 @@ describe("prepare arguments", function() {
     const functionDefinition: FunctionDefinition = {
       ndcKind: FunctionNdcKind.Function,
       description: null,
+      parallelDegree: null,
       arguments: [
         {
           argumentName: "nullOnlyArg",
@@ -314,6 +316,7 @@ describe("prepare arguments", function() {
     const functionDefinition: FunctionDefinition = {
       ndcKind: FunctionNdcKind.Function,
       description: null,
+      parallelDegree: null,
       arguments: [
         {
           argumentName: "stringArg",
@@ -396,6 +399,7 @@ describe("prepare arguments", function() {
     const functionDefinition: FunctionDefinition = {
       ndcKind: FunctionNdcKind.Function,
       description: null,
+      parallelDegree: null,
       arguments: [
         {
           argumentName: "dateTime",
@@ -449,6 +453,7 @@ describe("prepare arguments", function() {
     const functionDefinition: FunctionDefinition = {
       ndcKind: FunctionNdcKind.Function,
       description: null,
+      parallelDegree: null,
       arguments: [
         {
           argumentName: "literalString",