Merge pull request #84 from Arkatufus/Add_option_to_use_BIGINT_IDENTITY_for_journal_ordering_column

Add option to use BIGINT IDENTITY for journal ordering column

Author: Arkatufus

README.md

@@ -42,6 +42,19 @@ akka.persistence{
 
 			# defines column db type used to store payload. Available option: BYTEA (default), JSON, JSONB
 			stored-as = BYTEA
+
+			# Setting used to toggle sequential read access when loading large objects
+			# from journals and snapshot stores.
+			sequential-access = off
+
+			# When turned on, persistence will use `BIGINT` and `GENERATED ALWAYS AS IDENTITY`
+			# for journal table schema creation.
+			# NOTE: This only affects newly created tables, as such, it should not affect any
+			#       existing database.
+			#
+			# !!!!! WARNING !!!!!
+			# To use this feature, you have to have PorsgreSql version 10 or above
+			use-bigint-identity-for-ordering-column = off
 		}
 	}
 
@@ -71,6 +84,10 @@ akka.persistence{
 			
 			# defines column db type used to store payload. Available option: BYTEA (default), JSON, JSONB
 			stored-as = BYTEA
+
+			# Setting used to toggle sequential read access when loading large objects
+			# from journals and snapshot stores.
+			sequential-access = off
 		}
 	}
 }
@@ -110,6 +127,23 @@ CREATE TABLE {your_metadata_table_name} (
 );
 ```
 
+Note that if you turn on the `akka.persistence.journal.postgresql.use-bigint-identity-for-ordering-column` flag, the journal table schema will be altered to the latest recommended primary key setting.
+```
+CREATE TABLE {your_journal_table_name} (
+	ordering BIGINT NOT NULL GENERATED ALWAYS AS IDENTITY PRIMARY KEY,
+    persistence_id VARCHAR(255) NOT NULL,
+    sequence_nr BIGINT NOT NULL,
+    is_deleted BOOLEAN NOT NULL,
+    created_at BIGINT NOT NULL,
+    manifest VARCHAR(500) NOT NULL,
+    payload BYTEA NOT NULL,
+    tags VARCHAR(100) NULL,
+    serializer_id INTEGER NULL,
+    CONSTRAINT {your_journal_table_name}_uq UNIQUE (persistence_id, sequence_nr)
+);
+```
+Since this script is only run once during table generation, we will not provide any migration path for this change, any migration is left as an exercise for the user.
+
 ### Migration
 
 #### From 1.1.0 to 1.3.1

src/Akka.Persistence.PostgreSql.Tests/PostgreSqlJournalBigIntSpec.cs

@@ -0,0 +1,96 @@
+//-----------------------------------------------------------------------
+// <copyright file="PostgreSqlJournalSpec.cs" company="Akka.NET Project">
+//     Copyright (C) 2009-2016 Lightbend Inc. <http://www.lightbend.com>
+//     Copyright (C) 2013-2016 Akka.NET project <https://github.com/akkadotnet/akka.net>
+// </copyright>
+//-----------------------------------------------------------------------
+
+using System;
+using System.Collections.Immutable;
+using System.Linq;
+using System.Threading.Tasks;
+using Akka.Actor;
+using Akka.Configuration;
+using Akka.Persistence.TCK;
+using Akka.Persistence.TCK.Journal;
+using FluentAssertions;
+using Npgsql;
+using Xunit;
+using Xunit.Abstractions;
+
+namespace Akka.Persistence.PostgreSql.Tests
+{
+    [Collection("PostgreSqlSpec")]
+    public class PostgreSqlJournalBigIntSpec : JournalSpec
+    {
+        private static Config Initialize(PostgresFixture fixture) 
+        {
+            //need to make sure db is created before the tests start
+            DbUtils.Initialize(fixture);
+
+            var config = @"
+                akka.persistence {
+                    publish-plugin-commands = on
+                    journal {
+                        plugin = ""akka.persistence.journal.postgresql""
+                        postgresql {
+                            class = ""Akka.Persistence.PostgreSql.Journal.PostgreSqlJournal, Akka.Persistence.PostgreSql""
+                            plugin-dispatcher = ""akka.actor.default-dispatcher""
+                            table-name = event_journal
+                            schema-name = public
+                            auto-initialize = on
+                            connection-string = """ + DbUtils.ConnectionString + @"""
+                            use-bigint-identity-for-ordering-column = on
+                        }
+                    }
+                }
+                akka.test.single-expect-default = 10s";
+
+            return ConfigurationFactory.ParseString(config);
+        }
+
+        // TODO: hack. Replace when https://github.com/akkadotnet/akka.net/issues/3811
+        protected override bool SupportsSerialization => false;
+
+        public PostgreSqlJournalBigIntSpec(ITestOutputHelper output, PostgresFixture fixture)
+            : base(Initialize(fixture), "PostgreSqlJournalBigIntSpec", output)
+        {
+            Initialize();
+        }
+
+        protected override void Dispose(bool disposing)
+        {
+            base.Dispose(disposing);
+            DbUtils.Clean();
+        }
+
+        [Fact]
+        public async Task BigInt_Journal_ordering_column_data_type_should_be_BigInt()
+        {
+            using (var conn = new NpgsqlConnection(DbUtils.ConnectionString))
+            {
+                conn.Open();
+
+                var sql = $@"
+                SELECT column_name, column_default, data_type, is_identity, identity_generation
+                FROM information_schema.columns
+                WHERE table_schema = 'public'
+                    AND table_name = 'event_journal'
+                    AND ordinal_position = 1";
+
+                using (var cmd = new NpgsqlCommand(sql, conn))
+                {
+                    var reader = await cmd.ExecuteReaderAsync();
+                    await reader.ReadAsync();
+
+                    // these are the "fingerprint" of BIGINT ... GENERATED ALWAYS AS IDENTITY
+                    reader.GetString(0).Should().Be("ordering");
+                    reader[1].Should().BeOfType<DBNull>();
+                    reader.GetString(2).Should().Be("bigint");
+                    reader.GetString(3).Should().Be("YES");
+                    reader.GetString(4).Should().Be("ALWAYS");
+                }
+            }
+        }
+    }
+}

src/Akka.Persistence.PostgreSql.Tests/PostgreSqlJournalSpec.cs

@@ -5,8 +5,15 @@
 // </copyright>
 //-----------------------------------------------------------------------
 
+using System;
+using System.Reflection;
+using System.Threading.Tasks;
+using Akka.Actor;
 using Akka.Configuration;
+using Akka.Persistence.PostgreSql.Journal;
 using Akka.Persistence.TCK.Journal;
+using FluentAssertions;
+using Npgsql;
 using Xunit;
 using Xunit.Abstractions;
 
@@ -54,5 +61,34 @@ protected override void Dispose(bool disposing)
             base.Dispose(disposing);
             DbUtils.Clean();
         }
+
+        [Fact]
+        public async Task BigSerial_Journal_ordering_column_data_type_should_be_BigSerial()
+        {
+            using (var conn = new NpgsqlConnection(DbUtils.ConnectionString))
+            {
+                conn.Open();
+
+                var sql = $@"
+                SELECT column_name, column_default, data_type, is_identity, identity_generation
+                FROM information_schema.columns
+                WHERE table_schema = 'public'
+                    AND table_name = 'event_journal'
+                    AND ordinal_position = 1";
+
+                using (var cmd = new NpgsqlCommand(sql, conn))
+                {
+                    var reader = await cmd.ExecuteReaderAsync();
+                    await reader.ReadAsync();
+
+                    // these are the "fingerprint" of BIGSERIAL
+                    reader.GetString(0).Should().Be("ordering");
+                    reader.GetString(1).Should().Be("nextval('event_journal_ordering_seq'::regclass)");
+                    reader.GetString(2).Should().Be("bigint");
+                    reader.GetString(3).Should().Be("NO");
+                    reader[4].Should().BeOfType<DBNull>();
+                }
+            }
+        }
     }
 }

src/Akka.Persistence.PostgreSql/Journal/PostgreSqlJournal.cs

@@ -48,7 +48,8 @@ public PostgreSqlJournal(Config journalConfig) : base(journalConfig)
                 timeout: config.GetTimeSpan("connection-timeout"),
                 storedAs: storedAs,
                 defaultSerializer: config.GetString("serializer"),
-                useSequentialAccess: config.GetBoolean("sequential-access")),
+                useSequentialAccess: config.GetBoolean("sequential-access"),
+                useBigIntPrimaryKey: config.GetBoolean("use-bigint-identity-for-ordering-column")),
                     Context.System.Serialization,
                     GetTimestampProvider(config.GetString("timestamp-provider")));
 

src/Akka.Persistence.PostgreSql/Journal/PostgreSqlQueryExecutor.cs

@@ -34,9 +34,9 @@ public PostgreSqlQueryExecutor(PostgreSqlQueryConfiguration configuration, Akka.
         {
             var storedAs = configuration.StoredAs.ToString().ToUpperInvariant();
 
-            CreateEventsJournalSql = $@"
+            CreateEventsJournalSql =  $@"
                 CREATE TABLE IF NOT EXISTS {Configuration.FullJournalTableName} (
-                    {Configuration.OrderingColumnName} BIGSERIAL NOT NULL PRIMARY KEY,
+                    {Configuration.OrderingColumnName} {(configuration.UseBigIntPrimaryKey ? "BIGINT GENERATED ALWAYS AS IDENTITY" : "BIGSERIAL")} NOT NULL PRIMARY KEY,
                     {Configuration.PersistenceIdColumnName} VARCHAR(255) NOT NULL,
                     {Configuration.SequenceNrColumnName} BIGINT NOT NULL,
                     {Configuration.IsDeletedColumnName} BOOLEAN NOT NULL,
@@ -46,8 +46,7 @@ public PostgreSqlQueryExecutor(PostgreSqlQueryConfiguration configuration, Akka.
                     {Configuration.TagsColumnName} VARCHAR(100) NULL,
                     {Configuration.SerializerIdColumnName} INTEGER NULL,
                     CONSTRAINT {Configuration.JournalEventsTableName}_uq UNIQUE ({Configuration.PersistenceIdColumnName}, {Configuration.SequenceNrColumnName})
-                );
-                ";
+                );";
 
             CreateMetaTableSql = $@"
                 CREATE TABLE IF NOT EXISTS {Configuration.FullMetaTableName} (
@@ -254,6 +253,7 @@ public class PostgreSqlQueryConfiguration : QueryConfiguration
     {
         public readonly StoredAsType StoredAs;
         public readonly JsonSerializerSettings JsonSerializerSettings;
+        public readonly bool UseBigIntPrimaryKey;
 
         public PostgreSqlQueryConfiguration(
             string schemaName,
@@ -270,14 +270,16 @@ public PostgreSqlQueryConfiguration(
             string serializerIdColumnName,
             TimeSpan timeout,
             StoredAsType storedAs,
-            string defaultSerializer,
+            string defaultSerializer, 
             JsonSerializerSettings jsonSerializerSettings = null, 
-            bool useSequentialAccess = true)
+            bool useSequentialAccess = true, 
+            bool useBigIntPrimaryKey = false)
             : base(schemaName, journalEventsTableName, metaTableName, persistenceIdColumnName, sequenceNrColumnName,
                   payloadColumnName, manifestColumnName, timestampColumnName, isDeletedColumnName, tagsColumnName, orderingColumn, 
                 serializerIdColumnName, timeout, defaultSerializer, useSequentialAccess)
         {
             StoredAs = storedAs;
+            UseBigIntPrimaryKey = useBigIntPrimaryKey;
             JsonSerializerSettings = jsonSerializerSettings ?? new JsonSerializerSettings
             {
                 ContractResolver = new AkkaContractResolver()

src/Akka.Persistence.PostgreSql/postgresql.conf

@@ -37,6 +37,15 @@
 			# Setting used to toggle sequential read access when loading large objects
 			# from journals and snapshot stores.
 			sequential-access = off
+
+			# When turned on, persistence will use `BIGINT` and `GENERATED ALWAYS AS IDENTITY`
+			# for journal table schema creation.
+			# NOTE: This only affects newly created tables, as such, it should not affect any
+			#       existing database.
+			#
+			# !!!!! WARNING !!!!!
+			# To use this feature, you have to have PorsgreSql version 10 or above
+			use-bigint-identity-for-ordering-column = off
 		}
 	}