perf(core): test PostgreSQL query pipelining

Author: dlhck

bun.lock

@@ -113,6 +113,9 @@ export function getDbConfig(): DataSourceOptions {
                 port: process.env.CI ? +(process.env.E2E_POSTGRES_PORT || 5432) : 5432,
                 username: 'vendure',
                 password: 'password',
+                extra: {
+                    pipeline: true,
+                },
             };
         case 'mariadb':
             return {

e2e-common/test-config.ts

@@ -1,4 +1,4 @@
-import { mergeConfig } from '@vendure/core';
+import { mergeConfig, TransactionalConnection } from '@vendure/core';
 import { createErrorResultGuard, createTestEnvironment, ErrorResultGuard } from '@vendure/testing';
 import { fail } from 'assert';
 import path from 'path';
@@ -41,6 +41,18 @@ describe('Transaction infrastructure', () => {
         await server.destroy();
     });
 
+    itIfDb(['postgres'])('enables query pipelining on PostgreSQL connections', async () => {
+        const connection = server.app.get(TransactionalConnection).rawConnection;
+        const queryRunner = connection.createQueryRunner();
+
+        try {
+            const databaseConnection = await queryRunner.connect();
+            expect(databaseConnection.pipeline).toBe(true);
+        } finally {
+            await queryRunner.release();
+        }
+    });
+
     it('non-failing mutation', async () => {
         const { createTestAdministrator } = await adminClient.query(createTestAdministratorDocument, {
             emailAddress: 'test1',

packages/core/e2e/database-transactions.e2e-spec.ts

@@ -0,0 +1,99 @@
+/* eslint-disable no-console */
+import { Bench } from 'tinybench';
+import { DataSource, QueryRunner } from 'typeorm';
+import { PostgresConnectionOptions } from 'typeorm/driver/postgres/PostgresConnectionOptions';
+import { afterAll, beforeAll, describe, expect, it } from 'vitest';
+
+import { getDbConfig } from '../../../e2e-common/test-config';
+
+const BATCH_SIZE = 10;
+const expectedValues = Array.from({ length: BATCH_SIZE }, (_, index) => index);
+
+describe.skipIf(process.env.DB !== 'postgres')('PostgreSQL query pipelining benchmark', () => {
+    let sequentialDataSource: DataSource;
+    let pipelinedDataSource: DataSource;
+    let sequentialQueryRunner: QueryRunner;
+    let pipelinedQueryRunner: QueryRunner;
+
+    beforeAll(async () => {
+        const connectionOptions = getDbConfig();
+        if (connectionOptions.type !== 'postgres') {
+            throw new Error('This benchmark requires DB=postgres');
+        }
+
+        sequentialDataSource = await createDataSource(connectionOptions, false);
+        pipelinedDataSource = await createDataSource(connectionOptions, true);
+        sequentialQueryRunner = sequentialDataSource.createQueryRunner();
+        pipelinedQueryRunner = pipelinedDataSource.createQueryRunner();
+        await Promise.all([sequentialQueryRunner.connect(), pipelinedQueryRunner.connect()]);
+    });
+
+    afterAll(async () => {
+        await Promise.all([sequentialQueryRunner?.release(), pipelinedQueryRunner?.release()]);
+        await Promise.all([sequentialDataSource?.destroy(), pipelinedDataSource?.destroy()]);
+    });
+
+    it('compares batches of concurrent queries on one TypeORM QueryRunner', async () => {
+        const sequentialResult = await runBatch(sequentialQueryRunner);
+        const pipelinedResult = await runBatch(pipelinedQueryRunner);
+        expect(getValues(sequentialResult)).toEqual(expectedValues);
+        expect(getValues(pipelinedResult)).toEqual(expectedValues);
+
+        const bench = new Bench({
+            warmupTime: 500,
+            time: 2000,
+        });
+        bench
+            .add('pipeline off', () => runBatch(sequentialQueryRunner))
+            .add('pipeline on', () => runBatch(pipelinedQueryRunner));
+
+        const tasks = await bench.run();
+        const sequentialQps = getQueriesPerSecond(tasks[0].result?.hz);
+        const pipelinedQps = getQueriesPerSecond(tasks[1].result?.hz);
+        const speedup = pipelinedQps / sequentialQps;
+
+        console.table([
+            { mode: 'pipeline off', queriesPerSecond: Math.round(sequentialQps) },
+            { mode: 'pipeline on', queriesPerSecond: Math.round(pipelinedQps) },
+            { mode: 'ratio', queriesPerSecond: `${speedup.toFixed(2)}x` },
+        ]);
+
+        expect(sequentialQps).toBeGreaterThan(0);
+        expect(pipelinedQps).toBeGreaterThan(0);
+    });
+});
+
+async function createDataSource(
+    connectionOptions: PostgresConnectionOptions,
+    pipeline: boolean,
+): Promise<DataSource> {
+    const dataSource = new DataSource({
+        ...connectionOptions,
+        database: 'postgres',
+        entities: [],
+        synchronize: false,
+        extra: {
+            ...connectionOptions.extra,
+            pipeline,
+        },
+    });
+    await dataSource.initialize();
+    return dataSource;
+}
+
+function runBatch(queryRunner: QueryRunner): Promise<any[]> {
+    return Promise.all(
+        expectedValues.map(value => queryRunner.query('SELECT $1::integer AS value', [value])),
+    );
+}
+
+function getValues(results: any[]): number[] {
+    return results.map(rows => rows[0].value);
+}
+
+function getQueriesPerSecond(iterationsPerSecond: number | undefined): number {
+    if (!iterationsPerSecond) {
+        throw new Error('Benchmark did not produce a throughput result');
+    }
+    return iterationsPerSecond * BATCH_SIZE;
+}

packages/core/e2e/postgres-pipelining.bench.ts

@@ -28,6 +28,7 @@
         "e2e": "cross-env PACKAGE=core vitest --config ../../e2e-common/vitest.config.mts --run",
         "e2e:watch": "cross-env PACKAGE=core vitest --config ../../e2e-common/vitest.config.mts",
         "bench": "cross-env PACKAGE=core vitest --config ../../e2e-common/vitest.config.bench.ts --run",
+        "bench:pg-pipeline": "cross-env PACKAGE=core DB=postgres vitest --config ../../e2e-common/vitest.config.bench.ts --run e2e/postgres-pipelining.bench.ts --reporter=verbose",
         "i18n:extract": "node ./scripts/translate/i18n-tool.mjs extract",
         "i18n:apply": "node ./scripts/translate/i18n-tool.mjs apply",
         "ci": "npm run build"
@@ -99,7 +100,7 @@
         "glob": "^10.3.10",
         "ioredis": "^5.3.2",
         "mysql2": "^3.15.0",
-        "pg": "^8.13.1",
+        "pg": "^8.23.0",
         "rimraf": "^5.0.5",
         "sql.js": "1.13.0",
         "typescript": "5.8.2"

packages/core/package.json

@@ -34,6 +34,9 @@ export function getPostgresConnectionOptions(databaseName: string) {
         username: 'admin',
         password: 'secret',
         database: databaseName,
+        extra: {
+            pipeline: true,
+        },
     };
 }
 

packages/dev-server/load-testing/load-test-config.ts

@@ -52,7 +52,7 @@
         "@types/pg": "^8.11.2",
         "@vendure/core": "3.7.2",
         "mysql2": "^3.15.0",
-        "pg": "^8.11.3",
+        "pg": "^8.23.0",
         "rimraf": "^5.0.5",
         "typescript": "5.8.2"
     }