Batch concurrent requests serially

Author: kmcginnes

Changelog.md

@@ -2,6 +2,12 @@
 
 ## Upcoming
 
+**Major Changes**
+
+- Requests executed in parallel for each label or class are now batched in
+  groups of 10 to reduce chance of throttling errors
+  (<https://github.com/aws/graph-explorer/pull/489>)
+
 **Bug Fixes and Minor Changes**
 
 - Increase request body size limit for proxy server

packages/graph-explorer/src/connector/openCypher/queries/fetchSchema.test.ts

@@ -400,10 +400,10 @@ describe("OpenCypher > fetchSchema", () => {
     await fetchSchema(openCypherFetchFn);
 
     expect(openCypherFetchFn.mock.calls[3][0]).toStrictEqual(
-      "MATCH() -[e:`route`]- () RETURN e AS object LIMIT 1"
+      "MATCH () -[e:`route`]- () RETURN e AS object LIMIT 1"
     );
     expect(openCypherFetchFn.mock.calls[4][0]).toStrictEqual(
-      "MATCH() -[e:`contains`]- () RETURN e AS object LIMIT 1"
+      "MATCH () -[e:`contains`]- () RETURN e AS object LIMIT 1"
     );
   });
 
@@ -422,10 +422,10 @@ describe("OpenCypher > fetchSchema", () => {
     await fetchSchema(openCypherFetchFn);
 
     expect(openCypherFetchFn.mock.calls[3][0]).toStrictEqual(
-      "MATCH() -[e:`route`]- () RETURN e AS object LIMIT 1"
+      "MATCH () -[e:`route`]- () RETURN e AS object LIMIT 1"
     );
     expect(openCypherFetchFn.mock.calls[4][0]).toStrictEqual(
-      "MATCH() -[e:`contains`]- () RETURN e AS object LIMIT 1"
+      "MATCH () -[e:`contains`]- () RETURN e AS object LIMIT 1"
     );
   });
 });

packages/graph-explorer/src/connector/openCypher/queries/fetchSchema.ts

@@ -1,5 +1,10 @@
-import { sanitizeText } from "../../../utils";
-import type { SchemaResponse } from "../../useGEFetchTypes";
+import { batchPromisesSerially, sanitizeText } from "../../../utils";
+import { DEFAULT_CONCURRENT_REQUESTS_LIMIT } from "../../../utils/constants";
+import type {
+  EdgeSchemaResponse,
+  SchemaResponse,
+  VertexSchemaResponse,
+} from "../../useGEFetchTypes";
 import edgeLabelsTemplate from "../templates/edgeLabelsTemplate";
 import edgesSchemaTemplate from "../templates/edgesSchemaTemplate";
 import vertexLabelsTemplate from "../templates/vertexLabelsTemplate";
@@ -27,11 +32,14 @@ type RawEdgeLabelsResponse = {
 };
 
 type RawVerticesSchemaResponse = {
-  results: [
-    {
-      object: OCVertex;
-    },
-  ];
+  results:
+    | [
+        {
+          object: OCVertex;
+        },
+      ]
+    | []
+    | undefined;
 };
 
 type RawEdgesSchemaResponse = {
@@ -41,7 +49,8 @@ type RawEdgesSchemaResponse = {
           object: OCEdge;
         },
       ]
-    | [];
+    | []
+    | undefined;
 };
 
 // Fetches all vertex labels and their counts
@@ -79,29 +88,39 @@ const fetchVerticesAttributes = async (
   labels: Array<string>,
   countsByLabel: Record<string, number>
 ): Promise<SchemaResponse["vertices"]> => {
-  const vertices: SchemaResponse["vertices"] = [];
-
   if (labels.length === 0) {
-    return vertices;
+    return [];
   }
 
-  await Promise.all(
-    labels.map(async labelResult => {
+  const responses = await batchPromisesSerially(
+    labels,
+    DEFAULT_CONCURRENT_REQUESTS_LIMIT,
+    async label => {
       const verticesTemplate = verticesSchemaTemplate({
-        type: labelResult,
+        type: label,
       });
 
       const response =
         await openCypherFetch<RawVerticesSchemaResponse>(verticesTemplate);
 
-      const vertex = response.results[0]?.object as OCVertex;
-      if (!vertex) {
-        return;
+      return {
+        vertex: response.results ? response.results[0]?.object : null,
+        label,
+      };
+    }
+  );
+
+  const vertices = responses
+    .map(({ vertex }) => {
+      // verify response has the info we need
+      if (!vertex || !vertex["~labels"]) {
+        return null;
       }
 
-      const label = vertex["~labels"][0] as string;
+      // Use the first label
+      const label = vertex["~labels"][0];
       const properties = vertex["~properties"];
-      vertices.push({
+      const vertexSchema: VertexSchemaResponse = {
         type: label,
         displayLabel: sanitizeText(label),
         total: countsByLabel[label],
@@ -113,9 +132,10 @@ const fetchVerticesAttributes = async (
             dataType: typeof value === "string" ? "String" : "Number",
           };
         }),
-      });
+      };
+      return vertexSchema;
     })
-  );
+    .filter(vertexSchema => vertexSchema != null);
 
   return vertices;
 };
@@ -170,44 +190,41 @@ const fetchEdgesAttributes = async (
   labels: Array<string>,
   countsByLabel: Record<string, number>
 ): Promise<SchemaResponse["edges"]> => {
-  const edges: SchemaResponse["edges"] = [];
-
   if (labels.length === 0) {
-    return edges;
+    return [];
   }
 
-  await Promise.all(
-    labels.map(async labelResult => {
+  const responses = await batchPromisesSerially(
+    labels,
+    DEFAULT_CONCURRENT_REQUESTS_LIMIT,
+    async label => {
       const edgesTemplate = edgesSchemaTemplate({
-        type: labelResult,
+        type: label,
       });
 
       const response =
         await openCypherFetch<RawEdgesSchemaResponse>(edgesTemplate);
 
-      // verify response has the info we need
-      if (
-        !response.results ||
-        response.results.length === 0 ||
-        !response.results[0].object
-      ) {
-        return;
-      }
-
-      const edge = response.results[0].object as OCEdge;
-      const entityType = edge["~entityType"] as string;
-      const type = edge["~type"] as string;
+      return {
+        edge: response.results ? response.results[0]?.object : null,
+        label,
+      };
+    }
+  );
 
+  const edges = responses
+    .map(({ edge, label }) => {
       // verify response has the info we need
-      if (!entityType || !type) {
-        return;
+      if (!edge || !edge["~entityType"] || !edge["~type"]) {
+        return null;
       }
 
+      const type = edge["~type"];
       const properties = edge["~properties"];
-      edges.push({
+      const edgeSchema: EdgeSchemaResponse = {
         type: type,
         displayLabel: sanitizeText(type),
-        total: countsByLabel[labelResult],
+        total: countsByLabel[label],
         attributes: Object.entries(properties || {}).map(([name, prop]) => {
           const value = prop;
           return {
@@ -216,9 +233,10 @@ const fetchEdgesAttributes = async (
             dataType: typeof value === "string" ? "String" : "Number",
           };
         }),
-      });
+      };
+      return edgeSchema;
     })
-  );
+    .filter(edgeSchema => edgeSchema != null);
 
   return edges;
 };

packages/graph-explorer/src/connector/openCypher/templates/edgeSchemaTemplate.test.ts

@@ -5,7 +5,7 @@ describe("OpenCypher > edgesSchemaTemplate", () => {
     const template = edgesSchemaTemplate({ type: "route" });
 
     expect(template).toBe(
-      `MATCH() -[e:\`route\`]- () RETURN e AS object LIMIT 1`
+      `MATCH () -[e:\`route\`]- () RETURN e AS object LIMIT 1`
     );
   });
 });

packages/graph-explorer/src/connector/openCypher/templates/edgesSchemaTemplate.ts

@@ -10,7 +10,7 @@
  * LIMIT 1`
  */
 const edgesSchemaTemplate = ({ type }: { type: string }) => {
-  return `MATCH() -[e:\`${type}\`]- () RETURN e AS object LIMIT 1`;
+  return `MATCH () -[e:\`${type}\`]- () RETURN e AS object LIMIT 1`;
 };
 
 export default edgesSchemaTemplate;

packages/graph-explorer/src/connector/sparql/queries/fetchSchema.ts

@@ -1,3 +1,5 @@
+import { batchPromisesSerially } from "../../../utils";
+import { DEFAULT_CONCURRENT_REQUESTS_LIMIT } from "../../../utils/constants";
 import type { SchemaResponse } from "../../useGEFetchTypes";
 import classesWithCountsTemplates from "../templates/classesWithCountsTemplates";
 import predicatesByClassTemplate from "../templates/predicatesByClassTemplate";
@@ -62,11 +64,12 @@ const fetchPredicatesByClass = async (
   classes: Array<string>,
   countsByClass: Record<string, number>
 ) => {
-  const vertices: SchemaResponse["vertices"] = [];
-  await Promise.all(
-    classes.map(async classResult => {
+  const responses = await batchPromisesSerially(
+    classes,
+    DEFAULT_CONCURRENT_REQUESTS_LIMIT,
+    async resourceClass => {
       const classPredicatesTemplate = predicatesByClassTemplate({
-        class: classResult,
+        class: resourceClass,
       });
       const predicatesResponse =
         await sparqlFetch<RawPredicatesSamplesResponse>(
@@ -81,22 +84,25 @@ const fetchPredicatesByClass = async (
         dataType: TYPE_MAP[item.sample.datatype || ""] || "String",
       }));
 
-      vertices.push({
-        type: classResult,
-        displayLabel: "",
-        total: countsByClass[classResult],
-        displayNameAttribute:
-          attributes.find(attr => displayNameCandidates.includes(attr.name))
-            ?.name || "id",
-        longDisplayNameAttribute:
-          attributes.find(attr => displayDescCandidates.includes(attr.name))
-            ?.name || "types",
+      return {
         attributes,
-      });
-    })
+        resourceClass,
+      };
+    }
   );
 
-  return vertices;
+  return responses.map(({ attributes, resourceClass }) => ({
+    type: resourceClass,
+    displayLabel: "",
+    total: countsByClass[resourceClass],
+    displayNameAttribute:
+      attributes.find(attr => displayNameCandidates.includes(attr.name))
+        ?.name || "id",
+    longDisplayNameAttribute:
+      attributes.find(attr => displayDescCandidates.includes(attr.name))
+        ?.name || "types",
+    attributes,
+  }));
 };
 
 const fetchClassesSchema = async (sparqlFetch: SparqlFetch) => {

packages/graph-explorer/src/utils/batchPromisesSerially.ts

@@ -0,0 +1,28 @@
+import { chunk } from "lodash";
+
+/**
+ * Performs the given operation on the collection of items in groups of
+ * concurrent requests. A batch group must complete entirely before progressing
+ * to the next group.
+ *
+ * @param items The array of items that will be passed to the callback
+ * @param batchSize The size of the batch groups
+ * @param callback The async operation to perform on the given item
+ * @returns The results of all the operations performed on the given items in
+ * the order of the items array.
+ */
+export default async function batchPromisesSerially<Item, Result>(
+  items: Item[],
+  batchSize: number,
+  callback: (item: Item) => Promise<Result>
+): Promise<Result[]> {
+  const batches = chunk(items, batchSize);
+  const results = new Array<Result>();
+
+  for (const batch of batches) {
+    const batchResults = await Promise.all(batch.map(callback));
+    results.push(...batchResults);
+  }
+
+  return results;
+}

packages/graph-explorer/src/utils/constants.ts

@@ -1,3 +1,4 @@
 export const DEFAULT_SERVICE_TYPE = "neptune-db";
 export const DEFAULT_FETCH_TIMEOUT = 240000;
 export const DEFAULT_NODE_EXPAND_LIMIT = 500;
+export const DEFAULT_CONCURRENT_REQUESTS_LIMIT = 10;

packages/graph-explorer/src/utils/index.ts

@@ -7,5 +7,6 @@ export { default as useClickOutside } from "./useClickOutside";
 export { default as sanitizeText } from "./sanitizeText";
 export { DEFAULT_SERVICE_TYPE } from "./constants";
 export { default as escapeString } from "./escapeString";
+export { default as batchPromisesSerially } from "./batchPromisesSerially";
 export * from "./set";
 export * from "./env";