Skip to content

Commit 19c0a1c

Browse files
kmcginneskmcginnes
authored
Batch openCypher schema-sync attribute sampling into UNION ALL requests (#2099)
openCypher schema discovery issued one HTTP request per label — thousands of requests per sync on graphs with many labels. Sample attributes for a whole batch of labels in a single UNION ALL of index-scoped per-label blocks, chunked into DEFAULT_BATCH_REQUEST_SIZE groups and run through the concurrency pool. Edge blocks use a directed match (-[e]->) so each type is scanned once. The UNION ALL of per-label MATCH blocks is the only index-backed shape portable across every supported Neptune engine (1.2.1.0-1.4.7.0). SchemaResponse output is unchanged.
1 parent a4b2429 commit 19c0a1c

6 files changed

Lines changed: 184 additions & 115 deletions

File tree

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

Lines changed: 16 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -1,11 +1,23 @@
1+
import { normalize } from "@/utils/testing";
2+
13
import edgesSchemaTemplate from "./edgesSchemaTemplate";
24

35
describe("OpenCypher > edgesSchemaTemplate", () => {
4-
it("Should return a template with the projection of each type", () => {
5-
const template = edgesSchemaTemplate({ type: "route" });
6+
it("returns one directed sample block per edge type, joined by UNION ALL", () => {
7+
const template = edgesSchemaTemplate({ types: ["route", "contains"] });
8+
9+
expect(normalize(template)).toBe(
10+
normalize(`
11+
MATCH () -[e:\`route\`]-> () RETURN e AS object LIMIT 1
12+
UNION ALL
13+
MATCH () -[e:\`contains\`]-> () RETURN e AS object LIMIT 1
14+
`),
15+
);
16+
});
617

7-
expect(template).toBe(
8-
`MATCH () -[e:\`route\`]- () RETURN e AS object LIMIT 1`,
18+
it("returns a single block for a single type", () => {
19+
expect(edgesSchemaTemplate({ types: ["route"] })).toBe(
20+
"MATCH () -[e:`route`]-> () RETURN e AS object LIMIT 1",
921
);
1022
});
1123
});

‎packages/graph-explorer/src/connector/openCypher/fetchSchema/edgesSchemaTemplate.ts‎

Lines changed: 19 additions & 14 deletions
Original file line numberDiff line numberDiff line change
@@ -1,18 +1,23 @@
1+
import { query } from "@/utils";
2+
3+
import { fragment } from "../fragments";
4+
15
/**
2-
* Given an edge type, it returns an OpenCypher template that contains
3-
* one sample of the edge.
6+
* Given a set of edge types, returns an openCypher query that samples one edge
7+
* per type in a single request: a `UNION ALL` of type-scoped blocks. The match
8+
* is directed (`-[e]->`) so each type is scanned once; an undirected `-[e]-`
9+
* would scan both directions and add a self-loop filter for the same sample.
410
*
511
* @example
6-
* type = "route"
7-
*
8-
* MATCH() -[e:`route`]- ()
9-
* RETURN e AS object
10-
* LIMIT 1`
12+
* edgesSchemaTemplate({ types: ["route", "contains"] })
13+
* // MATCH () -[e:`route`]-> () RETURN e AS object LIMIT 1
14+
* // UNION ALL
15+
* // MATCH () -[e:`contains`]-> () RETURN e AS object LIMIT 1
1116
*/
12-
import { fragment } from "../fragments";
13-
14-
const edgesSchemaTemplate = ({ type }: { type: string }) => {
15-
return `MATCH () -[e:${fragment.identifier(type)}]- () RETURN e AS object LIMIT 1`;
16-
};
17-
18-
export default edgesSchemaTemplate;
17+
export default function edgesSchemaTemplate({ types }: { types: string[] }) {
18+
const blocks = types.map(
19+
type =>
20+
`MATCH () -[e:${fragment.identifier(type)}]-> () RETURN e AS object LIMIT 1`,
21+
);
22+
return query`${blocks.join("\nUNION ALL\n")}`;
23+
}

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

Lines changed: 62 additions & 17 deletions
Original file line numberDiff line numberDiff line change
@@ -4,6 +4,7 @@ import type { SchemaResponse } from "@/connector/useGEFetchTypes";
44

55
import { ClientLoggerConnector } from "@/connector/LoggerConnector";
66
import { createEdgeType, createVertexType } from "@/core";
7+
import { normalize } from "@/utils/testing";
78

89
import fetchSchema from ".";
910

@@ -14,8 +15,7 @@ describe("OpenCypher > fetchSchema", () => {
1415
.mockResolvedValueOnce(allVertexLabelsResponse)
1516
.mockResolvedValueOnce(airportPropertiesResponse)
1617
.mockResolvedValueOnce(allEdgesResponse)
17-
.mockResolvedValueOnce(routeEdgePropertiesResponse)
18-
.mockResolvedValueOnce(containsEdgePropertiesResponse)
18+
.mockResolvedValueOnce(batchedEdgePropertiesResponse)
1919
.mockImplementation(query => {
2020
throw new Error(query);
2121
});
@@ -410,48 +410,85 @@ describe("OpenCypher > fetchSchema", () => {
410410
expect(schema.edges.length).toBe(1);
411411
});
412412

413-
it("Should request properties for edges where labels are strings", async () => {
413+
it("Should sample all edge types in a single batched request when labels are strings", async () => {
414414
const openCypherFetchFn = vi
415415
.fn()
416416
.mockResolvedValueOnce(allVertexLabelsResponse)
417417
.mockResolvedValueOnce(airportPropertiesResponse)
418418
.mockResolvedValueOnce(allEdgesResponse)
419-
.mockResolvedValueOnce(routeEdgePropertiesResponse)
420-
.mockResolvedValueOnce(containsEdgePropertiesResponse)
419+
.mockResolvedValueOnce(batchedEdgePropertiesResponse)
421420
.mockImplementation(query => {
422421
throw new Error(query);
423422
});
424423

425424
await fetchSchema(openCypherFetchFn, new ClientLoggerConnector());
426425

427-
expect(openCypherFetchFn.mock.calls[3][0]).toStrictEqual(
428-
"MATCH () -[e:`route`]- () RETURN e AS object LIMIT 1",
429-
);
430-
expect(openCypherFetchFn.mock.calls[4][0]).toStrictEqual(
431-
"MATCH () -[e:`contains`]- () RETURN e AS object LIMIT 1",
426+
expect(normalize(openCypherFetchFn.mock.calls[3][0])).toBe(
427+
normalize(`
428+
MATCH () -[e:\`route\`]-> () RETURN e AS object LIMIT 1
429+
UNION ALL
430+
MATCH () -[e:\`contains\`]-> () RETURN e AS object LIMIT 1
431+
`),
432432
);
433433
});
434434

435-
it("Should request properties for edges where labels are arrays of strings", async () => {
435+
it("Should sample all edge types in a single batched request when labels are arrays of strings", async () => {
436436
const openCypherFetchFn = vi
437437
.fn()
438438
.mockResolvedValueOnce(allVertexLabelsResponse)
439439
.mockResolvedValueOnce(airportPropertiesResponse)
440440
.mockResolvedValueOnce(allEdgesLabelsInArrayResponse)
441-
.mockResolvedValueOnce(routeEdgePropertiesResponse)
442-
.mockResolvedValueOnce(containsEdgePropertiesResponse)
441+
.mockResolvedValueOnce(batchedEdgePropertiesResponse)
443442
.mockImplementation(query => {
444443
throw new Error(query);
445444
});
446445

447446
await fetchSchema(openCypherFetchFn, new ClientLoggerConnector());
448447

449-
expect(openCypherFetchFn.mock.calls[3][0]).toStrictEqual(
450-
"MATCH () -[e:`route`]- () RETURN e AS object LIMIT 1",
448+
expect(normalize(openCypherFetchFn.mock.calls[3][0])).toBe(
449+
normalize(`
450+
MATCH () -[e:\`route\`]-> () RETURN e AS object LIMIT 1
451+
UNION ALL
452+
MATCH () -[e:\`contains\`]-> () RETURN e AS object LIMIT 1
453+
`),
451454
);
452-
expect(openCypherFetchFn.mock.calls[4][0]).toStrictEqual(
453-
"MATCH () -[e:`contains`]- () RETURN e AS object LIMIT 1",
455+
});
456+
457+
it("Should batch attribute sampling into one request per DEFAULT_BATCH_REQUEST_SIZE labels", async () => {
458+
const labelCount = 250;
459+
const vertexLabels = Array.from(
460+
{ length: labelCount },
461+
(_, i) => `Vertex${i}`,
454462
);
463+
const edgeLabels = Array.from({ length: labelCount }, (_, i) => `Edge${i}`);
464+
465+
const openCypherFetchFn = vi
466+
.fn()
467+
.mockResolvedValueOnce({
468+
results: vertexLabels.map(label => ({ label, count: 1 })),
469+
})
470+
.mockResolvedValueOnce({ results: [] })
471+
.mockResolvedValueOnce({ results: [] })
472+
.mockResolvedValueOnce({ results: [] })
473+
.mockResolvedValueOnce({
474+
results: edgeLabels.map(label => ({ label, count: 1 })),
475+
})
476+
.mockResolvedValueOnce({ results: [] })
477+
.mockResolvedValueOnce({ results: [] })
478+
.mockResolvedValueOnce({ results: [] })
479+
.mockImplementation(query => {
480+
throw new Error(query);
481+
});
482+
483+
await fetchSchema(openCypherFetchFn, new ClientLoggerConnector());
484+
485+
// 250 labels / 100 per batch = 3 attribute requests each, plus the two
486+
// label-count queries = 8 requests total (not 250 + 250 + 2).
487+
expect(openCypherFetchFn).toHaveBeenCalledTimes(8);
488+
489+
const firstVertexBatch = openCypherFetchFn.mock.calls[1][0] as string;
490+
expect(firstVertexBatch.match(/LIMIT 1/g)).toHaveLength(100);
491+
expect(firstVertexBatch).toContain("UNION ALL");
455492
});
456493
});
457494

@@ -548,3 +585,11 @@ const containsEdgePropertiesResponse = {
548585
},
549586
],
550587
};
588+
589+
// One batched request returns a sample per edge type in a single `results` array.
590+
const batchedEdgePropertiesResponse = {
591+
results: [
592+
routeEdgePropertiesResponse.results[0],
593+
containsEdgePropertiesResponse.results[0],
594+
],
595+
};

‎packages/graph-explorer/src/connector/openCypher/fetchSchema/index.ts‎

Lines changed: 49 additions & 71 deletions
Original file line numberDiff line numberDiff line change
@@ -1,3 +1,5 @@
1+
import { chunk } from "lodash";
2+
13
import type { LoggerConnector } from "@/connector/LoggerConnector";
24
import type { SchemaResponse } from "@/connector/useGEFetchTypes";
35

@@ -7,8 +9,11 @@ import {
79
mapEdgeToTypeConfig,
810
mapVertexToTypeConfigs,
911
} from "@/core";
10-
import { mapWithConcurrency } from "@/utils";
11-
import { DEFAULT_CONCURRENT_REQUESTS_LIMIT } from "@/utils/constants";
12+
import {
13+
mapWithConcurrency,
14+
DEFAULT_BATCH_REQUEST_SIZE,
15+
DEFAULT_CONCURRENT_REQUESTS_LIMIT,
16+
} from "@/utils";
1217

1318
import type { OCEdge, OCVertex } from "../types";
1419
import type { GraphSummary, OpenCypherFetch } from "../types";
@@ -40,25 +45,11 @@ type RawEdgeLabelsResponse = {
4045
};
4146

4247
type RawVerticesSchemaResponse = {
43-
results:
44-
| [
45-
{
46-
object: OCVertex;
47-
},
48-
]
49-
| []
50-
| undefined;
48+
results?: Array<{ object: OCVertex }>;
5149
};
5250

5351
type RawEdgesSchemaResponse = {
54-
results:
55-
| [
56-
{
57-
object: OCEdge;
58-
},
59-
]
60-
| []
61-
| undefined;
52+
results?: Array<{ object: OCEdge }>;
6253
};
6354

6455
// Fetches all vertex labels and their counts
@@ -108,36 +99,30 @@ const fetchVerticesAttributes = async (
10899
}
109100

110101
remoteLogger.info("[openCypher Explorer] Fetching vertices attributes...");
111-
const responses = await mapWithConcurrency(
112-
labels,
102+
const batches = chunk(labels, DEFAULT_BATCH_REQUEST_SIZE);
103+
const batchResults = await mapWithConcurrency(
104+
batches,
113105
DEFAULT_CONCURRENT_REQUESTS_LIMIT,
114-
async label => {
115-
const verticesTemplate = verticesSchemaTemplate({
116-
type: label,
106+
async batch => {
107+
const response = await openCypherFetch<RawVerticesSchemaResponse>(
108+
verticesSchemaTemplate({ types: batch }),
109+
);
110+
111+
return (response.results ?? []).flatMap(({ object: ocVertex }) => {
112+
// verify response has the info we need
113+
if (!ocVertex || !ocVertex["~labels"]) {
114+
return [];
115+
}
116+
117+
const vertex = createVertex(mapApiVertex(ocVertex));
118+
return mapVertexToTypeConfigs(vertex).map(vertexTypeConfig => ({
119+
...vertexTypeConfig,
120+
total: countsByLabel[vertexTypeConfig.type],
121+
}));
117122
});
118-
119-
const response =
120-
await openCypherFetch<RawVerticesSchemaResponse>(verticesTemplate);
121-
122-
return response.results ? response.results[0]?.object : null;
123123
},
124124
);
125-
126-
const vertices = responses
127-
.flatMap(ocVertex => {
128-
// verify response has the info we need
129-
if (!ocVertex || !ocVertex["~labels"]) {
130-
return null;
131-
}
132-
133-
const vertex = createVertex(mapApiVertex(ocVertex));
134-
const vertexTypeConfigs = mapVertexToTypeConfigs(vertex);
135-
return vertexTypeConfigs.map(vertexTypeConfig => ({
136-
...vertexTypeConfig,
137-
total: countsByLabel[vertexTypeConfig.type],
138-
}));
139-
})
140-
.filter(vertexSchema => vertexSchema != null);
125+
const vertices = batchResults.flat();
141126

142127
remoteLogger.info(
143128
`[openCypher Explorer] Found ${vertices.flatMap(v => v.attributes).length} vertex attributes across ${vertices.length} vertex types.`,
@@ -214,37 +199,30 @@ const fetchEdgesAttributes = async (
214199
}
215200

216201
remoteLogger.info("[openCypher Explorer] Fetching edges attributes...");
217-
const responses = await mapWithConcurrency(
218-
labels,
202+
const batches = chunk(labels, DEFAULT_BATCH_REQUEST_SIZE);
203+
const batchResults = await mapWithConcurrency(
204+
batches,
219205
DEFAULT_CONCURRENT_REQUESTS_LIMIT,
220-
async label => {
221-
const edgesTemplate = edgesSchemaTemplate({
222-
type: label,
206+
async batch => {
207+
const response = await openCypherFetch<RawEdgesSchemaResponse>(
208+
edgesSchemaTemplate({ types: batch }),
209+
);
210+
211+
return (response.results ?? []).flatMap(({ object: ocEdge }) => {
212+
// verify response has the info we need
213+
if (!ocEdge || !ocEdge["~entityType"] || !ocEdge["~type"]) {
214+
return [];
215+
}
216+
217+
const edge = createEdge(mapApiEdge(ocEdge));
218+
const edgeTypeConfig = mapEdgeToTypeConfig(edge);
219+
return [
220+
{ ...edgeTypeConfig, total: countsByLabel[edgeTypeConfig.type] },
221+
];
223222
});
224-
225-
const response =
226-
await openCypherFetch<RawEdgesSchemaResponse>(edgesTemplate);
227-
228-
return response.results ? response.results[0]?.object : null;
229223
},
230224
);
231-
232-
const edges = responses
233-
.map(ocEdge => {
234-
// verify response has the info we need
235-
if (!ocEdge || !ocEdge["~entityType"] || !ocEdge["~type"]) {
236-
return null;
237-
}
238-
239-
const edge = createEdge(mapApiEdge(ocEdge));
240-
const edgeTypeConfig = mapEdgeToTypeConfig(edge);
241-
242-
return {
243-
...edgeTypeConfig,
244-
total: countsByLabel[edgeTypeConfig.type],
245-
};
246-
})
247-
.filter(edgeSchema => edgeSchema != null);
225+
const edges = batchResults.flat();
248226

249227
remoteLogger.info(
250228
`[openCypher Explorer] Found ${edges.flatMap(e => e.attributes).length} edge attributes across ${edges.length} edge types.`,

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

Lines changed: 18 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -1,15 +1,29 @@
1+
import { normalize } from "@/utils/testing";
2+
13
import { UnescapableValueError } from "../../queryValueError";
24
import verticesSchemaTemplate from "./verticesSchemaTemplate";
35

46
describe("OpenCypher > verticesSchemaTemplate", () => {
5-
it("Should return a template with the projection of each type", () => {
6-
const template = verticesSchemaTemplate({ type: "country" });
7+
it("returns one index-scoped sample block per label, joined by UNION ALL", () => {
8+
const template = verticesSchemaTemplate({ types: ["airport", "country"] });
79

8-
expect(template).toBe("MATCH (v:`country`) RETURN v AS object LIMIT 1");
10+
expect(normalize(template)).toBe(
11+
normalize(`
12+
MATCH (v:\`airport\`) RETURN v AS object LIMIT 1
13+
UNION ALL
14+
MATCH (v:\`country\`) RETURN v AS object LIMIT 1
15+
`),
16+
);
17+
});
18+
19+
it("returns a single block for a single label", () => {
20+
expect(verticesSchemaTemplate({ types: ["country"] })).toBe(
21+
"MATCH (v:`country`) RETURN v AS object LIMIT 1",
22+
);
923
});
1024

1125
it("throws on a control-character label rather than emitting a malformed query", () => {
12-
expect(() => verticesSchemaTemplate({ type: "coun\ntry" })).toThrow(
26+
expect(() => verticesSchemaTemplate({ types: ["coun\ntry"] })).toThrow(
1327
UnescapableValueError,
1428
);
1529
});

0 commit comments

Comments
 (0)