Tag table with backwards compatibility

Akka.Persistence.Linq2Db.sln

@@ -43,6 +43,12 @@ Project("{2150E333-8FDC-42A3-9474-1A3956D46DE8}") = "build-system", "build-syste
  build-system\windows-release.yaml = build-system\windows-release.yaml
  EndProjectSection
 EndProject
+Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Akka.Linq2Db.Sandbox", "src\Akka.Linq2Db.Sandbox\Akka.Linq2Db.Sandbox.csproj", "{C9D2CCA5-D431-44F7-B8FC-5424655291B6}"
+EndProject
+Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Akka.Persistence.Linq2Db.IndexHelperApp", "src\Akka.Persistence.Linq2Db.IndexHelperApp\Akka.Persistence.Linq2Db.IndexHelperApp.csproj", "{7B03FDD2-45CC-4F83-B23F-66B4EE82437A}"
+EndProject
+Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Akka.Persistence.Linq2Db.IndexHelperLib", "src\Akka.Persistence.Linq2Db.IndexHelperLib\Akka.Persistence.Linq2Db.IndexHelperLib.csproj", "{FD78AE77-C95F-4D5D-90EC-923935BAC580}"
+EndProject
 Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Akka.Persistence.Linq2Db.Data.Compatibility.Tests", "src\Akka.Persistence.Linq2Db.Data.Compatibility.Tests\Akka.Persistence.Linq2Db.Data.Compatibility.Tests.csproj", "{5240D81C-0F6C-4F4A-85A1-6442C6D6ECBF}"
 EndProject
 Project("{2150E333-8FDC-42A3-9474-1A3956D46DE8}") = "Sql Scripts", "Sql Scripts", "{C4AC49FF-9ECF-4D38-A201-1105BBF7E628}"
@@ -113,6 +119,18 @@ Global
  {170698FA-DA1E-40BC-896D-AFA67976C0EB}.Debug|Any CPU.Build.0 = Debug|Any CPU
  {170698FA-DA1E-40BC-896D-AFA67976C0EB}.Release|Any CPU.ActiveCfg = Release|Any CPU
  {170698FA-DA1E-40BC-896D-AFA67976C0EB}.Release|Any CPU.Build.0 = Release|Any CPU
+ {C9D2CCA5-D431-44F7-B8FC-5424655291B6}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
+ {C9D2CCA5-D431-44F7-B8FC-5424655291B6}.Debug|Any CPU.Build.0 = Debug|Any CPU
+ {C9D2CCA5-D431-44F7-B8FC-5424655291B6}.Release|Any CPU.ActiveCfg = Release|Any CPU
+ {C9D2CCA5-D431-44F7-B8FC-5424655291B6}.Release|Any CPU.Build.0 = Release|Any CPU
+ {7B03FDD2-45CC-4F83-B23F-66B4EE82437A}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
+ {7B03FDD2-45CC-4F83-B23F-66B4EE82437A}.Debug|Any CPU.Build.0 = Debug|Any CPU
+ {7B03FDD2-45CC-4F83-B23F-66B4EE82437A}.Release|Any CPU.ActiveCfg = Release|Any CPU
+ {7B03FDD2-45CC-4F83-B23F-66B4EE82437A}.Release|Any CPU.Build.0 = Release|Any CPU
+ {FD78AE77-C95F-4D5D-90EC-923935BAC580}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
+ {FD78AE77-C95F-4D5D-90EC-923935BAC580}.Debug|Any CPU.Build.0 = Debug|Any CPU
+ {FD78AE77-C95F-4D5D-90EC-923935BAC580}.Release|Any CPU.ActiveCfg = Release|Any CPU
+ {FD78AE77-C95F-4D5D-90EC-923935BAC580}.Release|Any CPU.Build.0 = Release|Any CPU
  {5240D81C-0F6C-4F4A-85A1-6442C6D6ECBF}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
  {5240D81C-0F6C-4F4A-85A1-6442C6D6ECBF}.Debug|Any CPU.Build.0 = Debug|Any CPU
  {5240D81C-0F6C-4F4A-85A1-6442C6D6ECBF}.Release|Any CPU.ActiveCfg = Release|Any CPU

README.md

@@ -57,7 +57,26 @@ Working:
  - Classes used in place of ValueTuples in certain areas
  - We don't have separate Query classes at this time. This can definitely be improved in future
  - A couple of places around `WriteMessagesAsync` have had their logic moved to facilitate performance (i.e. use of `await` instead of `ContinueWith`)
- - Backwards Compatibility mode is implemented, to interoperate with existing journals and snapsho stores.
+ - Backwards Compatibility mode is implemented, to interoperate with existing journals and snapshot stores.
+
+- Tag Table Support (Alpha):
+ - Allows the writing of tags to a separate table to allow for different performance strategies when working with tags.
+ - Supports Two Tag Table Modes:
+ - WriteUUID: The tag table and join uses a 'sequential-uuid' type field that will have lower page splits while allowing for good row locality on insert.
+ - This option is intended for those who want maximum write performance, at the expense of database storage and load.
+ - OrderingId: Uses the Journal Row's 'ordering' sequential Int64 for the tag table and join.
+ - This option is intended for those who want more efficient use of the DB's space
+ - This will result in slower writes, but faster/more efficient reads.
+ - Provides multiple modes of operation for reads and writes, note that there are separate switches for both read and write!
+ - CommaSeparatedOnly: The old behavior, where the comma separated tags are held in a column
+ - CommaSeparatedAndTagTable: Will Read/Write from both the Comma Separated column as well as the Tag Table
+ - TagTableOnly: will only use the tag table for Read/Write
+ - 'Live' Migration should be possible via the following flow:
+ 1. Run Migration scripts to create new columns/tables.
+ 2. Rolling Deploy your system with Reads and Writes in 'CommaSeparatedAndTagTable' mode.
+ 3. Rolling deploy your system (again), with Writes now in 'TagTableOnly' mode.
+ 4. Run Migration App/Script to move existing tags into tag table.
+ 5. Rolling deploy your system (last one!) with Reads now in 'TagTableOnly' mode.
 
 ## Currently Implemented:
 

scripts/MySql/1_Migration_Setup.sql

@@ -1,13 +1,14 @@
-CREATE TABLE IF NOT EXISTS TagTable(
+CREATE TABLE IF NOT EXISTS tags(
  ordering_id BIGINT NOT NULL,
  tag NVARCHAR(64) NOT NULL,
  PRIMARY KEY (ordering_id, tag)
 );
 
 DROP PROCEDURE IF EXISTS Split;
-DELIMITER #
+
+DELIMITER ??
 CREATE PROCEDURE Split()
-proc_main: BEGIN
+BEGIN
 
  DECLARE v_cursor_done TINYINT UNSIGNED DEFAULT 0;
  DECLARE Id INT UNSIGNED;
@@ -38,7 +39,7 @@ proc_main: BEGIN
  END IF;
 
  IF LENGTH(slice) > 0 THEN
- INSERT IGNORE INTO TagTable (ordering_id, tag) VALUES (Id, slice);
+ INSERT IGNORE INTO tags (ordering_id, tag) VALUES (Id, slice);
  END IF;
 
  SET String = RIGHT(String, LENGTH(String) - idx);
@@ -51,5 +52,5 @@ proc_main: BEGIN
 
  CLOSE v_cursor;
 
-END proc_main #
+END??
 DELIMITER ;

scripts/PostgreSql/1_Migration_Setup.sql

@@ -1,4 +1,4 @@
-CREATE TABLE IF NOT EXISTS "public"."TagTable"(
+CREATE TABLE IF NOT EXISTS "public"."tags"(
  ordering_id BIGINT NOT NULL,
  tag VARCHAR(64) NOT NULL,
  PRIMARY KEY (ordering_id, tag)
@@ -10,7 +10,7 @@ BEGIN
  FOR var_t IN(SELECT unnest(string_to_array(tags, ';')) AS t) 
  LOOP 
  CONTINUE WHEN var_t.t IS NULL OR var_t.t = '';
- INSERT INTO "public"."TagTable" (ordering_id, tag) 
+ INSERT INTO "public"."tags" (ordering_id, tag) 
  VALUES (id, var_t.t)
  ON CONFLICT DO NOTHING;
  END LOOP;

scripts/SqlServer/1_Migration_Setup.sql

@@ -1,7 +1,7 @@
 IF NOT EXISTS(SELECT * FROM INFORMATION_SCHEMA.TABLES
- WHERE TABLE_SCHEMA = N'dbo' AND TABLE_NAME = N'TagTable')
+ WHERE TABLE_SCHEMA = N'dbo' AND TABLE_NAME = N'tags')
 BEGIN
- CREATE TABLE [dbo].[TagTable](
+ CREATE TABLE [dbo].[tags](
  ordering_id BIGINT NOT NULL,
  tag NVARCHAR(64) NOT NULL,
  PRIMARY KEY (ordering_id, tag)

scripts/SqlServer/2_Migration.sql

@@ -1,10 +1,10 @@
-INSERT INTO [dbo].[TagTable]([ordering_id], [tag])
+INSERT INTO [dbo].[tags]([ordering_id], [tag])
  SELECT * FROM (
  SELECT a.[Ordering], b.[items] FROM
  [dbo].[EventJournal] AS a
  CROSS APPLY [dbo].[Split](a.Tags, ';') b
  ) AS s([ordering_id], [tag])
  WHERE NOT EXISTS (
- SELECT * FROM [dbo].[TagTable] t WITH (updlock)
+ SELECT * FROM [dbo].[tags] t WITH (updlock)
  WHERE s.[ordering_id] = t.[ordering_id] AND s.[tag] = t.[tag]
  );

src/Akka.Linq2Db.Sandbox/Akka.Linq2Db.Sandbox.csproj

@@ -0,0 +1,14 @@
+<Project Sdk="Microsoft.NET.Sdk">
+
+ <PropertyGroup>
+ <TargetFramework>netcoreapp3.1</TargetFramework>
+ <Nullable>enable</Nullable>
+ </PropertyGroup>
+
+ <ItemGroup>
+ <PackageReference Include="Akka.Streams" Version="1.4.21" />
+ <PackageReference Include="linq2db" Version="3.4.1" />
+ <PackageReference Include="Reactive.Streams" Version="1.0.2" />
+ </ItemGroup>
+
+</Project>

src/Akka.Persistence.Linq2Db.Benchmark.DockerComparisonTests/SqlCommon/DockerBatchingSqlServerJournalPerfSpec.cs

@@ -17,7 +17,7 @@ public DockerBatchingSqlServerJournalPerfSpec(ITestOutputHelper output, SqlServe
  public static Config InitConfig(SqlServerFixture fixture)
  {
  //need to make sure db is created before the tests start
- DockerDbUtils.Initialize(fixture.ConnectionString);
+ SqlServerDbUtils.Initialize(fixture.ConnectionString);
  var specString = $@"
 akka.persistence {{
  publish-plugin-commands = on
@@ -29,7 +29,7 @@ class = ""Akka.Persistence.SqlServer.Journal.BatchingSqlServerJournal, Akka.Pers
  table-name = EventJournal
  schema-name = dbo
  auto-initialize = on
- connection-string = ""{DockerDbUtils.ConnectionString}""
+ connection-string = ""{SqlServerDbUtils.ConnectionString}""
  }}
  }}
 }}";

src/Akka.Persistence.Linq2Db.Benchmark.DockerComparisonTests/SqlCommon/DockerSqlServerJournalPerfSpec.cs

@@ -16,7 +16,7 @@ public DockerSqlServerJournalPerfSpec(ITestOutputHelper output, SqlServerFixture
  public static Config InitConfig(SqlServerFixture fixture)
  {
  //need to make sure db is created before the tests start
- DockerDbUtils.Initialize(fixture.ConnectionString);
+ SqlServerDbUtils.Initialize(fixture.ConnectionString);
  var specString = $@"
 akka.persistence {{
  publish-plugin-commands = on
@@ -28,7 +28,7 @@ class = ""Akka.Persistence.SqlServer.Journal.SqlServerJournal, Akka.Persistence.
  table-name = EventJournal
  schema-name = dbo
  auto-initialize = on
- connection-string = ""{DockerDbUtils.ConnectionString}""
+ connection-string = ""{SqlServerDbUtils.ConnectionString}""
  }}
  }}
 }}";

src/Akka.Persistence.Linq2Db.Benchmark.DockerTests/Docker/Linq2Db/DockerLinq2DbPostgreSQLJournalPerfSpec.cs

@@ -46,9 +46,8 @@ public DockerLinq2DbPostgreSqlJournalPerfSpec(ITestOutputHelper output,
  PostgreSqlFixture fixture) : base(InitConfig(fixture),
  "postgresperf", output,40, eventsCount: TestConstants.DockerNumMessages)
  {
- var extension = Linq2DbPersistence.Get(Sys);
- var config = Create(DockerDbUtils.ConnectionString)
- .WithFallback(extension.DefaultConfig)
+ var config = Create(SqlServerDbUtils.ConnectionString)
+ .WithFallback(Linq2DbPersistence.DefaultConfiguration)
  .GetConfig("akka.persistence.journal.linq2db");
  var connFactory = new AkkaPersistenceDataConnectionFactory(new JournalConfig(config));
  using var conn = connFactory.GetConnection();

src/Akka.Persistence.Linq2Db.Benchmark.DockerTests/Docker/Linq2Db/DockerLinq2DbSqlServerJournalPerfSpec.cs

@@ -49,9 +49,8 @@ public DockerLinq2DbSqlServerJournalPerfSpec(ITestOutputHelper output,
  SqlServerFixture fixture) : base(InitConfig(fixture),
  "sqlserverperf", output,40, eventsCount: TestConstants.DockerNumMessages)
  {
- var extension = Linq2DbPersistence.Get(Sys);
- var config = Create(DockerDbUtils.ConnectionString)
- .WithFallback(extension.DefaultConfig)
+ var config = Create(SqlServerDbUtils.ConnectionString)
+ .WithFallback(Linq2DbPersistence.DefaultConfiguration)
  .GetConfig("akka.persistence.journal.linq2db");
  var connFactory = new AkkaPersistenceDataConnectionFactory(new JournalConfig(config));
  using var conn = connFactory.GetConnection();
@@ -68,14 +67,14 @@ public DockerLinq2DbSqlServerJournalPerfSpec(ITestOutputHelper output,
  public static Config InitConfig(SqlServerFixture fixture)
  {
  //need to make sure db is created before the tests start
- DbUtils.Initialize(fixture.ConnectionString);
+ SqlServerDbUtils.Initialize(fixture.ConnectionString);
 
- return Create(DbUtils.ConnectionString);
+ return Create(SqlServerDbUtils.ConnectionString);
  } 
  protected override void Dispose(bool disposing)
  {
  base.Dispose(disposing);
- DbUtils.Clean();
+ SqlServerDbUtils.Clean();
  }
 
  [Fact]

src/Akka.Persistence.Linq2Db.Benchmark.Tests/Linq2Db/MSSQLiteLinq2DbJournalPerfSpec.cs

@@ -34,13 +34,11 @@ public static void InitWalForFileDb()
  public MsSqliteLinq2DbJournalPerfSpec(ITestOutputHelper output)
  : base(SqLiteJournalSpecConfig.Create(ConnString, ProviderName.SQLiteMS), "SqliteJournalSpec", output,eventsCount: TestConstants.NumMessages)
  {
- var extension = Linq2DbPersistence.Get(Sys);
-
  HeldSqliteConnection.Open();
  //InitWALForFileDb();
  var conf = new JournalConfig(
  SqLiteJournalSpecConfig.Create(ConnString, ProviderName.SQLiteMS)
- .WithFallback(extension.DefaultConfig)
+ .WithFallback(Linq2DbPersistence.DefaultConfiguration)
  .GetConfig("akka.persistence.journal.linq2db"));
 
  var connFactory = new AkkaPersistenceDataConnectionFactory(conf);

src/Akka.Persistence.Linq2Db.Compatibility.DockerTests/Akka.Persistence.Linq2Db.Compatibility.DockerTests.csproj

@@ -2,13 +2,9 @@
  <Import Project="..\common.props" />
 
  <PropertyGroup>
-
  <IsPackable>false</IsPackable>
-
  <RootNamespace>Akka.Persistence.Linq2Db.CompatibilityTests.Docker</RootNamespace>
-
  <TargetFramework>netcoreapp3.1</TargetFramework>
-
  </PropertyGroup>
 
  <ItemGroup>
@@ -32,16 +28,4 @@
  <ProjectReference Include="..\Akka.Persistence.Sql.Linq2Db.Tests\Akka.Persistence.Sql.Linq2Db.Tests.csproj" />
  </ItemGroup>
 
- <ItemGroup>
- <Compile Remove="Postgres\**" />
- </ItemGroup>
-
- <ItemGroup>
- <EmbeddedResource Remove="Postgres\**" />
- </ItemGroup>
-
- <ItemGroup>
- <None Remove="Postgres\**" />
- </ItemGroup>
-
 </Project>