Original #68 PR - Tag table with backwards compatibility

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

@@ -0,0 +1,15 @@
+<Project Sdk="Microsoft.NET.Sdk">
+
+    <PropertyGroup>
+        <TargetFramework>netcoreapp3.1</TargetFramework>
+        <ImplicitUsings>enable</ImplicitUsings>
+        <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>

Akka.Persistence.Linq2Db.IndexHelperApp/Akka.Persistence.Linq2Db.IndexHelperApp.csproj

@@ -0,0 +1,25 @@
+<Project Sdk="Microsoft.NET.Sdk">
+
+    <PropertyGroup>
+        <OutputType>Exe</OutputType>
+        <TargetFramework>netcoreapp3.1</TargetFramework>
+    </PropertyGroup>
+
+    <ItemGroup>
+      <ProjectReference Include="..\Akka.Persistence.Linq2Db.IndexHelperLib\Akka.Persistence.Linq2Db.IndexHelperLib.csproj" />
+      <ProjectReference Include="..\src\Akka.Persistence.Sql.Linq2Db.Tests\Akka.Persistence.Sql.Linq2Db.Tests.csproj" />
+    </ItemGroup>
+
+    <ItemGroup>
+      <PackageReference Include="CommandLineParser" Version="2.8.0" />
+      <PackageReference Include="FluentMigrator.Runner" Version="3.3.1" />
+    </ItemGroup>
+
+    <ItemGroup>
+      <None Remove="example.hocon" />
+      <Content Include="example.hocon">
+        <CopyToOutputDirectory>Always</CopyToOutputDirectory>
+      </Content>
+    </ItemGroup>
+
+</Project>

Akka.Persistence.Linq2Db.IndexHelperApp/Options.cs

@@ -0,0 +1,22 @@
+using CommandLine;
+
+namespace Akka.Persistence.Linq2Db.IndexHelperApp
+{
+    public class Options
+    {
+        [Option('f',"file", Required=true, HelpText = "Specify the HOCON file to use")]
+        public string File { get; set; }
+
+        [Option('p',"path", Required = true, HelpText = "The Path to the Akka.Persistence.Linq2Db Config in the HOCON.")]
+        public string HoconPath { get; set; }
+
+        [Option("OrderingIdx", Required = true, Group = "IndexType", HelpText = "Generates the SQL Text for an Ordering index")]
+        public bool GenerateOrdering { get; set; }
+
+        [Option("PidSeqNoIdx", Required = true, Group = "IndexType", HelpText = "Generates the SQL Text for an index on PersistenceID and SeqNo")]
+        public bool GeneratePidSeqNo { get; set; } 
+
+        [Option("TimeStampIdx", Required = true, Group = "IndexType", HelpText = "Generates the SQL Text for a Timestamp Index")]
+        public bool GenerateTimestamp { get; set; }
+    }
+}

Akka.Persistence.Linq2Db.IndexHelperApp/Program.cs

@@ -0,0 +1,141 @@
+using System;
+using System.IO;
+using Akka.Configuration;
+using Akka.Persistence.Linq2Db.IndexHelperLib;
+using Akka.Persistence.Sql.Linq2Db.Config;
+using Akka.Persistence.Sql.Linq2Db.Tests;
+using CommandLine;
+using FluentMigrator.Expressions;
+using FluentMigrator.Runner.Generators;
+using FluentMigrator.Runner.Generators.Generic;
+using FluentMigrator.Runner.Generators.MySql;
+using FluentMigrator.Runner.Generators.Oracle;
+using FluentMigrator.Runner.Generators.Postgres;
+using FluentMigrator.Runner.Generators.Postgres92;
+using FluentMigrator.Runner.Generators.SQLite;
+using FluentMigrator.Runner.Generators.SqlServer;
+using FluentMigrator.Runner.Processors.Postgres;
+using LinqToDB;
+using Microsoft.Extensions.Options;
+using Newtonsoft.Json;
+
+namespace Akka.Persistence.Linq2Db.IndexHelperApp
+{
+    class Program
+    {
+
+        static void Main(string[] args)
+        {
+            Parser.Default.ParseArguments<Options>(args)
+                .WithParsed(opts =>
+                {
+                    //var str = Linq2DbJournalDefaultSpecConfig.customConfig("testGen",
+                    //    "journalTbl", "metadataTbl", ProviderName.SqlServer,
+                    //    "connStr");
+                    var conf =
+                        ConfigurationFactory.ParseString(
+                            File.ReadAllText(opts.File));
+
+                    var journalConf =
+                        new Akka.Persistence.Sql.Linq2Db.Config.JournalConfig(
+                            conf.GetConfig(
+                                    opts.HoconPath
+                                    //"akka.persistence.journal.linq2db.testGen"
+                                    )
+                                .WithFallback(Akka.Persistence.Sql.Linq2Db
+                                    .Journal
+                                    .Linq2DbWriteJournal.DefaultConfiguration));
+                    var generator = getGenerator(journalConf.ProviderName);
+                    var helper = new JournalIndexHelper();
+                    CreateIndexExpression expr = null;
+                    GeneratePerOptions(opts, helper, journalConf, generator);
+                });
+        }
+
+        private static void GeneratePerOptions(Options opts, JournalIndexHelper helper,
+            JournalConfig journalConf, GenericGenerator generator)
+        {
+            CreateIndexExpression expr;
+            if (opts.GeneratePidSeqNo)
+            {
+                expr = new CreateIndexExpression()
+                {
+                    Index = helper.JournalOrdering(journalConf.TableConfig.TableName,
+                        journalConf.TableConfig.ColumnNames.Ordering,
+                        journalConf.TableConfig.SchemaName)
+                };
+                GenerateWithHeaderAndFooter(generator, expr, "Ordering");
+            }
+
+            if (opts.GeneratePidSeqNo)
+            {
+                expr = new CreateIndexExpression()
+                {
+                    Index = helper.DefaultJournalIndex(
+                        journalConf.TableConfig.TableName,
+                        journalConf.TableConfig.ColumnNames.PersistenceId,
+                        journalConf.TableConfig.ColumnNames.SequenceNumber,
+                        journalConf.TableConfig.SchemaName)
+                };
+                GenerateWithHeaderAndFooter(generator, expr, "PidAndSequenceNo");
+            }
+
+            if (opts.GenerateTimestamp)
+            {
+                expr = new CreateIndexExpression()
+                {
+                    Index = helper.JournalTimestamp(journalConf.TableConfig.TableName,
+                        journalConf.TableConfig.ColumnNames.Created,
+                        journalConf.TableConfig.SchemaName)
+                };
+                GenerateWithHeaderAndFooter(generator, expr, "Timestamp");
+            }
+        }
+
+        private static void GenerateWithHeaderAndFooter(GenericGenerator generator,
+            CreateIndexExpression expr, string indexType)
+        {
+            Console.WriteLine("-------");
+            Console.WriteLine($"----{indexType} Index Create Below");
+            Console.WriteLine(generator.Generate(expr));
+            Console.WriteLine($"----{indexType} Index Create Above");
+            Console.WriteLine("-------");
+        }
+
+        static GenericGenerator getGenerator(string dbArg)
+        {
+            if (dbArg.StartsWith("sqlserver",
+                StringComparison.InvariantCultureIgnoreCase))
+            {
+                return new SqlServer2008Generator();
+            }
+            else if (dbArg.Contains("sqlite",
+                StringComparison.InvariantCultureIgnoreCase))
+            {
+                return new SQLiteGenerator();
+            }
+            else if (dbArg.Contains("postgres",
+                StringComparison.InvariantCultureIgnoreCase))
+            {
+                return new Postgres92Generator(
+                    new PostgresQuoter(new PostgresOptions()),
+                    new OptionsWrapper<GeneratorOptions>(
+                        new GeneratorOptions()));
+            }
+            else if (dbArg.Contains("mysql",
+                StringComparison.InvariantCultureIgnoreCase))
+            {
+                return new MySql5Generator();
+            }
+            else if (dbArg.Contains("oracle",
+                StringComparison.InvariantCultureIgnoreCase))
+            {
+                return new OracleGenerator();
+            }
+            else
+            {
+                throw new Exception("IDK what to do with this!");
+            }
+        }
+    }
+}

Akka.Persistence.Linq2Db.IndexHelperApp/example.hocon

@@ -0,0 +1,15 @@
+akka.persistence.journal.linq2db{
+  testGen {
+  class = "Akka.Persistence.Sql.Linq2Db.Journal.Linq2DbWriteJournal, Akka.Persistence.Sql.Linq2Db"
+  provider-name = "SqlServer"
+    connection-string = "connStr"
+ tables{
+  journal{
+    auto-init = true
+    warn-on-auto-init-fail = false
+    table-name = "journalTbl"
+    metadata-table-name = "metadataTbl"
+  }
+ }
+}
+}

Akka.Persistence.Linq2Db.IndexHelperLib/Akka.Persistence.Linq2Db.IndexHelperLib.csproj

@@ -0,0 +1,15 @@
+<Project Sdk="Microsoft.NET.Sdk">
+
+    <PropertyGroup>
+        <TargetFramework>netstandard2.0</TargetFramework>
+    </PropertyGroup>
+
+    <ItemGroup>
+      <ProjectReference Include="..\src\Akka.Persistence.Sql.Linq2Db\Akka.Persistence.Sql.Linq2Db.csproj" />
+    </ItemGroup>
+
+    <ItemGroup>
+      <PackageReference Include="FluentMigrator" Version="3.3.1" />
+    </ItemGroup>
+
+</Project>

Akka.Persistence.Linq2Db.IndexHelperLib/Class1.cs

@@ -0,0 +1,50 @@
+using System;
+using FluentMigrator.Model;
+
+namespace Akka.Persistence.Linq2Db.IndexHelperLib
+{
+    public class JournalIndexHelper
+    {
+        public IndexDefinition DefaultJournalIndex(string tableName, string persistenceIdCol, string sequenceNoCol, string schemaName = null)
+        {
+            var idx =  beginCreateIndex(tableName, schemaName, $"UX_{tableName}_PID_SEQNO");
+            //short name for easy compat with all dbs. (*cough* oracle *cough*)
+            idx.Columns.Add(new IndexColumnDefinition(){ Name = persistenceIdCol });
+            idx.Columns.Add(new IndexColumnDefinition(){Name = sequenceNoCol, Direction = Direction.Ascending});
+            idx.IsUnique = true;
+            return idx;
+        }
+
+        public IndexDefinition JournalOrdering(string tableName,
+            string orderingCol, string schemaName = null)
+        {
+            var idx =  beginCreateIndex(tableName, schemaName,$"IX_{tableName}_Ordering");
+            idx.Columns.Add(new IndexColumnDefinition(){Name = orderingCol});
+            //Should it be?
+            //idx.IsUnique = true;
+            return idx;
+        }
+
+        public IndexDefinition JournalTimestamp(string tableName,
+            string timestampCol, string schemaName = null)
+        {
+            var idx = beginCreateIndex(tableName, schemaName,
+                $"IX_{tableName}_TimeStamp");
+            idx.Columns.Add(new IndexColumnDefinition(){Name = timestampCol});
+            //Not unique by any stretch.
+            return idx;
+        }
+
+        private static IndexDefinition beginCreateIndex(string tableName, string schemaName, string indexName)
+        {
+            var idx = new IndexDefinition();
+            if (string.IsNullOrWhiteSpace(schemaName) == false)
+            {
+                idx.SchemaName = schemaName;
+            }
+            idx.TableName = tableName;
+            idx.Name = indexName;
+            return idx;
+        }
+    }
+}

Akka.Persistence.Linq2Db.sln

@@ -36,6 +36,12 @@ EndProjectSection
 EndProject
 Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Akka.Persistence.Linq2Db.Benchmark.DockerComparisonTests", "Akka.Persistence.Linq2Db.Benchmark.DockerComparisonTests\Akka.Persistence.Linq2Db.Benchmark.DockerComparisonTests.csproj", "{170698FA-DA1E-40BC-896D-AFA67976C0EB}"
 EndProject
+Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Akka.Persistence.Linq2Db.IndexHelperLib", "Akka.Persistence.Linq2Db.IndexHelperLib\Akka.Persistence.Linq2Db.IndexHelperLib.csproj", "{AACE3FBC-51FE-4A9B-B6C4-4CCA750DB22E}"
+EndProject
+Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Akka.Persistence.Linq2Db.IndexHelperApp", "Akka.Persistence.Linq2Db.IndexHelperApp\Akka.Persistence.Linq2Db.IndexHelperApp.csproj", "{D5C851AA-DB80-4E9F-BD2E-03E63DC3082E}"
+EndProject
+Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Akka.Linq2Db.Sandbox", "Akka.Linq2Db.Sandbox\Akka.Linq2Db.Sandbox.csproj", "{697B9FC8-29E2-4F7D-B63B-9E4B873F6AA1}"
+EndProject
 Global
 	GlobalSection(SolutionConfigurationPlatforms) = preSolution
 		Debug|Any CPU = Debug|Any CPU
@@ -78,6 +84,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
+		{AACE3FBC-51FE-4A9B-B6C4-4CCA750DB22E}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
+		{AACE3FBC-51FE-4A9B-B6C4-4CCA750DB22E}.Debug|Any CPU.Build.0 = Debug|Any CPU
+		{AACE3FBC-51FE-4A9B-B6C4-4CCA750DB22E}.Release|Any CPU.ActiveCfg = Release|Any CPU
+		{AACE3FBC-51FE-4A9B-B6C4-4CCA750DB22E}.Release|Any CPU.Build.0 = Release|Any CPU
+		{D5C851AA-DB80-4E9F-BD2E-03E63DC3082E}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
+		{D5C851AA-DB80-4E9F-BD2E-03E63DC3082E}.Debug|Any CPU.Build.0 = Debug|Any CPU
+		{D5C851AA-DB80-4E9F-BD2E-03E63DC3082E}.Release|Any CPU.ActiveCfg = Release|Any CPU
+		{D5C851AA-DB80-4E9F-BD2E-03E63DC3082E}.Release|Any CPU.Build.0 = Release|Any CPU
+		{697B9FC8-29E2-4F7D-B63B-9E4B873F6AA1}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
+		{697B9FC8-29E2-4F7D-B63B-9E4B873F6AA1}.Debug|Any CPU.Build.0 = Debug|Any CPU
+		{697B9FC8-29E2-4F7D-B63B-9E4B873F6AA1}.Release|Any CPU.ActiveCfg = Release|Any CPU
+		{697B9FC8-29E2-4F7D-B63B-9E4B873F6AA1}.Release|Any CPU.Build.0 = Release|Any CPU
 	EndGlobalSection
 	GlobalSection(SolutionProperties) = preSolution
 		HideSolutionNode = FALSE

src/Akka.Persistence.Sql.Linq2Db/Config/BaseByteArrayJournalDaoConfig.cs

@@ -10,6 +10,7 @@ public BaseByteArrayJournalDaoConfig(Configuration.Config config)
             BufferSize = config.GetInt("buffer-size", 5000);
             BatchSize = config.GetInt("batch-size", 100);
             DbRoundTripBatchSize = config.GetInt("db-round-trip-max-batch-size", 1000);
+            DbRoundTripTagBatchSize = config.GetInt("db-round-trip-max-tag-batch-size", 1000);
             PreferParametersOnMultiRowInsert =
                 config.GetBoolean("prefer-parameters-on-multirow-insert",
                     false);
@@ -43,6 +44,6 @@ public BaseByteArrayJournalDaoConfig(Configuration.Config config)
         public int BufferSize { get; protected set; }
 
         public bool SqlCommonCompatibilityMode { get; protected set; }
-
+        public int DbRoundTripTagBatchSize { get; set; }
     }
 }

src/Akka.Persistence.Sql.Linq2Db/Config/JournalConfig.cs

@@ -21,7 +21,6 @@ public JournalConfig(Configuration.Config config)
             UseSharedDb = string.IsNullOrWhiteSpace(dbConf) ? null : dbConf;
             UseCloneConnection =
                 config.GetBoolean("use-clone-connection", false);
-
         }
 
         public string MaterializerDispatcher { get; protected set; }
@@ -44,6 +43,7 @@ public IDaoConfig IDaoConfig
         public string ProviderName { get; }
         public string ConnectionString { get; }
         public bool UseCloneConnection { get; set; }
+
     }
 
     public interface IProviderConfig<TTable>
@@ -61,4 +61,6 @@ public interface IDaoConfig
         bool SqlCommonCompatibilityMode { get; }
         int Parallelism { get; }
     }
+
+
 }