WIP [skip-ci]

src/Benchmark/Bench.cs

@@ -4,25 +4,42 @@
 using Fluss;
 using Fluss.Authentication;
 using Fluss.Events;
+using Fluss.PostgreSQL;
 using Fluss.ReadModel;
 using Microsoft.Extensions.DependencyInjection;
+using Npgsql;
 
 namespace Benchmark;
 
-[GitJob("879573d", baseline: true, id: "0_before")]
-[SimpleJob(id: "1_after")]
+// [GitJob("a4d91117", baseline: true, id: "0_before")]
+[SimpleJob(id: "current")]
 [RPlotExporter]
 [MemoryDiagnoser]
 public class Bench
 {
+    [Params("in-memory", "postgres")]
+    public string StorageType { get; set; }
+
     [IterationSetup]
     public void Setup()
     {
         var sc = new ServiceCollection();
         sc.AddEventSourcing();
+
+        if (StorageType == "postgres")
+        {
+            sc.AddPostgresEventSourcingRepository("Host=localhost;Port=5432;Database=fluss;Username=fluss;Password=fluss");
+        }
+
         sc.AddPolicy<AllowAllPolicy>();
 
         _sp = sc.BuildServiceProvider();
+
+        if (StorageType == "postgres") {
+            var migrator = _sp.GetRequiredService<Migrator>();
+            migrator.StartAsync(default).Wait();
+            migrator.WaitForFinish().Wait();
+        }
     }
 
     ServiceProvider _sp = null!;
@@ -51,14 +68,23 @@ await _sp.GetSystemUserUnitOfWorkFactory().Commit(async unitOfWork =>
         return sum;
     }
 
+    [IterationCleanup]
+    public void Cleanup()
+    {
+        if (StorageType == "postgres") {
+            var conn = new NpgsqlConnection("Host=localhost;Port=5432;Database=fluss;Username=fluss;Password=fluss");
+            conn.Open();
+            using var cmd = conn.CreateCommand();
+            cmd.CommandText = @"DELETE FROM ""Events""";
+            cmd.ExecuteNonQuery();
+            conn.Close();
+        }
+    }
+
     [IterationSetup(Targets = [nameof(PublishEventsAndReadReadHeavySingleReadModel), nameof(PublishEventsAndReadReadHeavyMultipleReadModel)])]
     public void SetupHeavyRead()
     {
-        var sc = new ServiceCollection();
-        sc.AddEventSourcing();
-        sc.AddPolicy<AllowAllPolicy>();
-
-        _sp = sc.BuildServiceProvider();
+        Setup();
 
         _sp.GetSystemUserUnitOfWorkFactory().Commit(async unitOfWork =>
         {

src/Benchmark/Benchmark.csproj

@@ -14,6 +14,7 @@
 
     <ItemGroup>
         <ProjectReference Include="..\Fluss\Fluss.csproj" />
+        <ProjectReference Include="..\Fluss.PostgreSQL\Fluss.PostgreSQL.csproj" />
     </ItemGroup>
 
 </Project>

src/Benchmark/Program.cs

@@ -1,4 +1,11 @@
 using Benchmark;
 using BenchmarkDotNet.Running;
 
-BenchmarkRunner.Run<Bench>();
+BenchmarkRunner.Run<Bench>();
+
+// var bench = new Bench();
+// bench.StorageType = "postgres";
+// bench.SetupHeavyRead();
+// await bench.PublishEventsAndReadReadHeavySingleReadModel();
+// bench.SetupHeavyRead();
+// bench.Cleanup();

src/Benchmark/docker-compose.yaml

@@ -0,0 +1,9 @@
+services:
+  db:
+    image: postgres:15
+    environment:
+      POSTGRES_USER: fluss
+      POSTGRES_PASSWORD: fluss
+      POSTGRES_DB: fluss
+    ports:
+      - "5432:5432"

src/Fluss.PostgreSQL/PostgreSQLEventRepository.cs

@@ -17,6 +17,10 @@ public partial class PostgreSQLEventRepository : IBaseEventRepository
     public PostgreSQLEventRepository(PostgreSQLConfig config)
     {
         var dataSourceBuilder = new NpgsqlDataSourceBuilder(config.ConnectionString);
+        dataSourceBuilder.ConnectionStringBuilder.Pooling = true;
+        dataSourceBuilder.ConnectionStringBuilder.MaxPoolSize = 100;
+        dataSourceBuilder.ConnectionStringBuilder.MinPoolSize = 1;
+        dataSourceBuilder.ConnectionStringBuilder.Enlist = false;
         dataSourceBuilder.UseJsonNet(settings: new JsonSerializerSettings
         {
             TypeNameHandling = TypeNameHandling.All,