Optimize event loading by single aggregateId (special case)

benchmarks/eventstore/save-event.ts

@@ -28,7 +28,7 @@ type EventSetSpecification = {
 }
 
 function autoAggregateId(i: number) {
-  return `AUTO_AGGREGATE_${i}}`
+  return `AUTO_AGGREGATE_${i}`
 }
 
 async function measureSaveEvent(

packages/runtime/adapters/eventstore-adapters/eventstore-postgresql/src/create-load-query.ts

@@ -7,13 +7,21 @@ type LoadFilter = Omit<CursorFilter, 'limit'> & {
   limit?: CursorFilter['limit']
 }
 
+const makeThreadArrayQuery = (vectorConditions: number[]) => {
+  return `${vectorConditions
+    .map(
+      (threadCounter, threadId) =>
+        `"threadId" = ${threadId} AND "threadCounter" >= ${threadCounter}::${INT8_SQL_TYPE} `
+    )
+    .join(' OR ')}`
+}
+
 const createLoadQuery = (
   { escapeId, escape, eventsTableName, databaseName }: AdapterPool,
   { eventTypes, aggregateIds, cursor, limit }: LoadFilter
 ) => {
   const vectorConditions = cursorToThreadArray(cursor)
   const injectString = (value: any): string => `${escape(value)}`
-  const injectNumber = (value: any): string => `${+value}`
 
   const queryConditions: string[] = ['1 = 1']
   if (eventTypes != null) {
@@ -35,6 +43,16 @@ const createLoadQuery = (
   const eventsTableAsId: string = escapeId(eventsTableName)
 
   if (limit !== undefined) {
+    if (aggregateIds != null && aggregateIds.length === 1) {
+      const resultVectorConditions = makeThreadArrayQuery(vectorConditions)
+      return [
+        `SELECT * FROM ${databaseNameAsId}.${eventsTableAsId}`,
+        `WHERE (${resultVectorConditions}) AND (${resultQueryCondition})`,
+        `ORDER BY "aggregateVersion" ASC`,
+        `LIMIT ${+limit}`,
+      ].join('\n')
+    }
+
     return [
       `SELECT "sortedEvents".* FROM (`,
       `  SELECT "unitedEvents".* FROM (`,
@@ -45,7 +63,7 @@ const createLoadQuery = (
               `SELECT * FROM ${databaseNameAsId}.${eventsTableAsId}`,
               `WHERE (${resultQueryCondition}) AND "threadId" = ${threadId} AND "threadCounter" >= ${threadCounter}::${INT8_SQL_TYPE}`,
               `ORDER BY "threadCounter" ASC`,
-              `LIMIT ${limit}`,
+              `LIMIT ${+limit}`,
             ].join(' ')})`
         )
         .join(' UNION ALL \n'),
@@ -58,14 +76,7 @@ const createLoadQuery = (
       `"sortedEvents"."threadId" ASC`,
     ].join('\n')
   } else {
-    const resultVectorConditions = `${vectorConditions
-      .map(
-        (threadCounter, threadId) =>
-          `"threadId" = ${injectNumber(
-            threadId
-          )} AND "threadCounter" >= ${threadCounter}::${INT8_SQL_TYPE} `
-      )
-      .join(' OR ')}`
+    const resultVectorConditions = makeThreadArrayQuery(vectorConditions)
 
     return [
       `SELECT * FROM ${databaseNameAsId}.${eventsTableAsId}`,

tests/eventstore-filter-events/index.test.ts

@@ -367,3 +367,75 @@ describe(`${adapterFactory.name}. Eventstore adapter events filtering`, () => {
     ).rejects.toThrow()
   })
 })
+
+describe(`${adapterFactory.name}. Eventstore adapter big aggregate`, () => {
+  beforeAll(adapterFactory.create('events_big_aggregate'))
+  afterAll(adapterFactory.destroy('events_big_aggregate'))
+
+  const adapter = adapters['events_big_aggregate']
+
+  const aggregateCount = 3
+  const eventsPerAggregate = 120
+  const loadStep = 50
+
+  test('should save many events with different aggregate ids', async () => {
+    const waitForMillisecond = async (event: StoredEvent) => {
+      if (Date.now() === event.timestamp) {
+        await new Promise((resolve) => {
+          setTimeout(resolve, 1)
+        })
+      }
+    }
+
+    for (let eventIndex = 0; eventIndex < eventsPerAggregate; eventIndex++) {
+      const event = {
+        aggregateId: `aggregateId`,
+        aggregateVersion: eventIndex + 1,
+        type: `EVENT`,
+        payload: null,
+        timestamp: 1,
+      }
+
+      for (
+        let aggregateIndex = 1;
+        aggregateIndex <= aggregateCount;
+        ++aggregateIndex
+      ) {
+        event.aggregateId = `aggregateId_${aggregateIndex}`
+        const saveResult = await adapter.saveEvent(event)
+        await waitForMillisecond(saveResult.event)
+      }
+    }
+
+    const { eventCount } = await adapter.describe()
+    expect(eventCount).toEqual(eventsPerAggregate * aggregateCount)
+  })
+
+  test('should consequentially load events by each aggregateId', async () => {
+    for (
+      let aggregateIndex = 1;
+      aggregateIndex <= aggregateCount;
+      ++aggregateIndex
+    ) {
+      const aggregateId = `aggregateId_${aggregateIndex}`
+      let currentCursor = null
+      let loadedEventCount = 0
+
+      for (let i = 0; i < eventsPerAggregate; i += loadStep) {
+        const {
+          events: loadedEvents,
+          cursor: nextCursor,
+        } = await adapter.loadEvents({
+          limit: loadStep,
+          cursor: currentCursor,
+          aggregateIds: [aggregateId],
+        })
+
+        loadedEventCount += loadedEvents.length
+        currentCursor = nextCursor
+      }
+
+      expect(loadedEventCount).toEqual(eventsPerAggregate)
+    }
+  })
+})