Skip to content

Commit 9e41724

Browse files
ianmacartneyianmacartney
committed
update manual testing
1 parent af5bfbf commit 9e41724

2 files changed

Lines changed: 147 additions & 117 deletions

File tree

‎example/convex/batchedWrites.ts‎

Lines changed: 30 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -10,7 +10,7 @@
1010
*/
1111

1212
import { DirectAggregate, TableAggregate } from "@convex-dev/aggregate";
13-
import { mutation } from "./_generated/server";
13+
import { internalMutation, mutation } from "./_generated/server";
1414
import { components } from "./_generated/api.js";
1515
import { v } from "convex/values";
1616
import { customMutation } from "convex-helpers/server/customFunctions";
@@ -32,21 +32,30 @@ const leaderboardAggregate = new TableAggregate<{
3232
sumValue: (doc) => doc.score,
3333
});
3434

35+
export const reset = internalMutation({
36+
args: {},
37+
handler: async (ctx) => {
38+
await aggregate.clearAll(ctx);
39+
},
40+
});
41+
3542
/**
3643
* Basic example: Enable buffering, queue operations, then flush manually.
3744
*/
38-
export const basicBatchedWrites = mutation({
45+
export const basicBatchedWrites = internalMutation({
3946
args: {
4047
count: v.number(),
4148
},
4249
handler: async (ctx, { count }) => {
4350
// Enable buffering mode - modifies the aggregate instance in place
4451
aggregate.startBuffering();
4552

53+
const initialCount = await aggregate.count(ctx);
54+
4655
// Queue multiple insert operations
4756
for (let i = 0; i < count; i++) {
4857
await aggregate.insert(ctx, {
49-
key: i,
58+
key: i + initialCount,
5059
id: `item-${i}`,
5160
sumValue: i * 10,
5261
});
@@ -58,6 +67,11 @@ export const basicBatchedWrites = mutation({
5867
// Read operations work normally (and auto-flush if needed)
5968
const total = await aggregate.count(ctx);
6069

70+
if (total !== initialCount + count) {
71+
console.log({ initialCount, count, total });
72+
throw new Error("Total count is incorrect");
73+
}
74+
6175
return { inserted: count, total };
6276
},
6377
});
@@ -121,12 +135,23 @@ export const addMultipleScores = mutationWithTriggers({
121135
),
122136
},
123137
handler: async (ctx, { scores }) => {
138+
const initialSumValue = await leaderboardAggregate.sum(ctx);
139+
124140
// Just insert into the table - the trigger automatically
125141
// updates the aggregate, and buffering batches all the updates
126142
for (const { name, score } of scores) {
127143
await ctx.db.insert("leaderboard", { name, score });
128144
}
129145

146+
const totalSumValue = await leaderboardAggregate.sum(ctx);
147+
148+
if (
149+
totalSumValue !==
150+
initialSumValue + scores.reduce((acc, { score }) => acc + score, 0)
151+
) {
152+
throw new Error("Total sum value is incorrect");
153+
}
154+
130155
return {
131156
inserted: scores.length,
132157
message: `Added ${scores.length} scores with batched aggregate updates`,
@@ -168,7 +193,7 @@ export const compareTriggersWithAndWithoutBatching = mutation({
168193
useBatching: v.boolean(),
169194
},
170195
handler: async (ctx, { count, useBatching }) => {
171-
const start = Date.now();
196+
console.time();
172197

173198
const customCtx = triggers.wrapDB(ctx);
174199
if (useBatching) {
@@ -194,12 +219,11 @@ export const compareTriggersWithAndWithoutBatching = mutation({
194219
}
195220
}
196221

197-
const duration = Date.now() - start;
222+
console.timeEnd();
198223

199224
return {
200225
method: useBatching ? "with batching" : "without batching",
201226
count,
202-
durationMs: duration,
203227
message: useBatching
204228
? `1 batched call to aggregate component`
205229
: `${count} individual calls to aggregate component`,

‎src/component/public.ts‎

Lines changed: 117 additions & 111 deletions
Original file line numberDiff line numberDiff line change
@@ -227,141 +227,147 @@ export const batch = mutation({
227227
},
228228
returns: v.null(),
229229
handler: async (ctx, { operations }) => {
230-
// Group operations by namespace to fetch each tree once
231-
const namespaceGroups = new Map<string, typeof operations>();
232-
for (const op of operations) {
233-
const namespace = "namespace" in op ? op.namespace : undefined;
230+
// Map to store trees for each namespace
231+
const treesMap = new Map<string, ReturnType<typeof getOrCreateTree>>();
232+
233+
// Helper function to get or create tree for a namespace
234+
const getTreeForNamespace = async (namespace: any) => {
234235
// Use a sentinel value for undefined namespace since JSON.stringify(undefined) returns undefined
235-
const key = namespace === undefined ? "__undefined__" : JSON.stringify(namespace);
236-
if (!namespaceGroups.has(key)) {
237-
namespaceGroups.set(key, []);
236+
const key =
237+
namespace === undefined ? "__undefined__" : JSON.stringify(namespace);
238+
if (!treesMap.has(key)) {
239+
treesMap.set(
240+
key,
241+
getOrCreateTree(ctx.db, namespace, DEFAULT_MAX_NODE_SIZE, true),
242+
);
238243
}
239-
namespaceGroups.get(key)!.push(op);
240-
}
244+
return await treesMap.get(key)!;
245+
};
241246

242-
// Process each namespace group
243-
for (const [namespaceKey, ops] of namespaceGroups.entries()) {
244-
const namespace = namespaceKey === "__undefined__" ? undefined : JSON.parse(namespaceKey);
245-
const tree = await getOrCreateTree(
246-
ctx.db,
247-
namespace,
248-
DEFAULT_MAX_NODE_SIZE,
249-
true,
250-
);
251-
252-
// Process operations in order
253-
for (const op of ops) {
254-
if (op.type === "insert") {
255-
await insertHandler(
256-
ctx,
257-
{
258-
key: op.key,
259-
value: op.value,
260-
summand: op.summand,
261-
namespace: op.namespace,
262-
},
263-
tree,
264-
);
265-
} else if (op.type === "delete") {
247+
// Process operations in order
248+
for (const op of operations) {
249+
if (op.type === "insert") {
250+
const tree = await getTreeForNamespace(op.namespace);
251+
await insertHandler(
252+
ctx,
253+
{
254+
key: op.key,
255+
value: op.value,
256+
summand: op.summand,
257+
namespace: op.namespace,
258+
},
259+
tree,
260+
);
261+
} else if (op.type === "delete") {
262+
const tree = await getTreeForNamespace(op.namespace);
263+
await deleteHandler(
264+
ctx,
265+
{
266+
key: op.key,
267+
namespace: op.namespace,
268+
},
269+
tree,
270+
);
271+
} else if (op.type === "replace") {
272+
// Handle delete from original namespace
273+
const deleteTree = await getTreeForNamespace(op.namespace);
274+
await deleteHandler(
275+
ctx,
276+
{
277+
key: op.currentKey,
278+
namespace: op.namespace,
279+
},
280+
deleteTree,
281+
);
282+
// Handle insert to new namespace (which might be different)
283+
const insertTree = await getTreeForNamespace(op.newNamespace);
284+
await insertHandler(
285+
ctx,
286+
{
287+
key: op.newKey,
288+
value: op.value,
289+
summand: op.summand,
290+
namespace: op.newNamespace,
291+
},
292+
insertTree,
293+
);
294+
} else if (op.type === "deleteIfExists") {
295+
const tree = await getTreeForNamespace(op.namespace);
296+
try {
266297
await deleteHandler(
267298
ctx,
268-
{
269-
key: op.key,
270-
namespace: op.namespace,
271-
},
299+
{ key: op.key, namespace: op.namespace },
272300
tree,
273301
);
274-
} else if (op.type === "replace") {
302+
} catch (e) {
303+
if (
304+
e instanceof ConvexError &&
305+
e.data?.code === "DELETE_MISSING_KEY"
306+
) {
307+
continue;
308+
}
309+
throw e;
310+
}
311+
} else if (op.type === "replaceOrInsert") {
312+
// Handle delete from original namespace
313+
const deleteTree = await getTreeForNamespace(op.namespace);
314+
try {
275315
await deleteHandler(
276316
ctx,
277317
{
278318
key: op.currentKey,
279319
namespace: op.namespace,
280320
},
281-
tree,
321+
deleteTree,
282322
);
283-
await insertHandler(
284-
ctx,
285-
{
286-
key: op.newKey,
287-
value: op.value,
288-
summand: op.summand,
289-
namespace: op.newNamespace,
290-
},
291-
tree,
292-
);
293-
} else if (op.type === "deleteIfExists") {
294-
try {
295-
await deleteHandler(
296-
ctx,
297-
{ key: op.key, namespace: op.namespace },
298-
tree,
299-
);
300-
} catch (e) {
301-
if (
302-
e instanceof ConvexError &&
303-
e.data?.code === "DELETE_MISSING_KEY"
304-
) {
305-
continue;
306-
}
323+
} catch (e) {
324+
if (
325+
!(e instanceof ConvexError && e.data?.code === "DELETE_MISSING_KEY")
326+
) {
307327
throw e;
308328
}
309-
} else if (op.type === "replaceOrInsert") {
310-
try {
311-
await deleteHandler(
312-
ctx,
313-
{
314-
key: op.currentKey,
315-
namespace: op.namespace,
316-
},
317-
tree,
318-
);
319-
} catch (e) {
320-
if (
321-
!(e instanceof ConvexError && e.data?.code === "DELETE_MISSING_KEY")
322-
) {
323-
throw e;
324-
}
325-
}
326-
await insertHandler(
327-
ctx,
328-
{
329-
key: op.newKey,
330-
value: op.value,
331-
summand: op.summand,
332-
namespace: op.newNamespace,
333-
},
334-
tree,
335-
);
336-
} else if (op.type === "insertIfDoesNotExist") {
337-
// insertIfDoesNotExist is implemented as replaceOrInsert
338-
try {
339-
await deleteHandler(
340-
ctx,
341-
{
342-
key: op.key,
343-
namespace: op.namespace,
344-
},
345-
tree,
346-
);
347-
} catch (e) {
348-
if (
349-
!(e instanceof ConvexError && e.data?.code === "DELETE_MISSING_KEY")
350-
) {
351-
throw e;
352-
}
353-
}
354-
await insertHandler(
329+
}
330+
// Handle insert to new namespace (which might be different)
331+
const insertTree = await getTreeForNamespace(op.newNamespace);
332+
await insertHandler(
333+
ctx,
334+
{
335+
key: op.newKey,
336+
value: op.value,
337+
summand: op.summand,
338+
namespace: op.newNamespace,
339+
},
340+
insertTree,
341+
);
342+
} else if (op.type === "insertIfDoesNotExist") {
343+
const tree = await getTreeForNamespace(op.namespace);
344+
// insertIfDoesNotExist is implemented as replaceOrInsert
345+
try {
346+
await deleteHandler(
355347
ctx,
356348
{
357349
key: op.key,
358-
value: op.value,
359-
summand: op.summand,
360350
namespace: op.namespace,
361351
},
362352
tree,
363353
);
354+
} catch (e) {
355+
if (
356+
!(e instanceof ConvexError && e.data?.code === "DELETE_MISSING_KEY")
357+
) {
358+
throw e;
359+
}
364360
}
361+
await insertHandler(
362+
ctx,
363+
{
364+
key: op.key,
365+
value: op.value,
366+
summand: op.summand,
367+
namespace: op.namespace,
368+
},
369+
tree,
370+
);
365371
}
366372
}
367373
},

0 commit comments

Comments
 (0)