From 90d21198e01dc676733cba57a58fabf4d5e690c7 Mon Sep 17 00:00:00 2001 From: David Cumps Date: Mon, 2 Mar 2020 19:36:47 +0100 Subject: [PATCH 1/2] use arrays and gin index to improve tag searching --- .../Journal/PostgreSqlQueryExecutor.cs | 99 ++++++++++++++++--- 1 file changed, 86 insertions(+), 13 deletions(-) mode change 100644 => 100755 src/Akka.Persistence.PostgreSql/Journal/PostgreSqlQueryExecutor.cs diff --git a/src/Akka.Persistence.PostgreSql/Journal/PostgreSqlQueryExecutor.cs b/src/Akka.Persistence.PostgreSql/Journal/PostgreSqlQueryExecutor.cs old mode 100644 new mode 100755 index dd030fc..f3c7844 --- a/src/Akka.Persistence.PostgreSql/Journal/PostgreSqlQueryExecutor.cs +++ b/src/Akka.Persistence.PostgreSql/Journal/PostgreSqlQueryExecutor.cs @@ -17,10 +17,12 @@ using System.Data; using System.Data.Common; using System.Linq; -using System.Text; namespace Akka.Persistence.PostgreSql.Journal { + using System.Threading; + using System.Threading.Tasks; + public class PostgreSqlQueryExecutor : AbstractQueryExecutor { private readonly PostgreSqlQueryConfiguration _configuration; @@ -32,7 +34,16 @@ public PostgreSqlQueryExecutor(PostgreSqlQueryConfiguration configuration, Akka. { _configuration = configuration; var storedAs = configuration.StoredAs.ToString().ToUpperInvariant(); - + + var allEventColumnNames = $@" + e.{Configuration.PersistenceIdColumnName} as PersistenceId, + e.{Configuration.SequenceNrColumnName} as SequenceNr, + e.{Configuration.TimestampColumnName} as Timestamp, + e.{Configuration.IsDeletedColumnName} as IsDeleted, + e.{Configuration.ManifestColumnName} as Manifest, + e.{Configuration.PayloadColumnName} as Payload, + e.{Configuration.SerializerIdColumnName} as SerializerId"; + CreateEventsJournalSql = $@" CREATE TABLE IF NOT EXISTS {Configuration.FullJournalTableName} ( {Configuration.OrderingColumnName} BIGSERIAL NOT NULL PRIMARY KEY, @@ -42,10 +53,13 @@ public PostgreSqlQueryExecutor(PostgreSqlQueryConfiguration configuration, Akka. {Configuration.TimestampColumnName} BIGINT NOT NULL, {Configuration.ManifestColumnName} VARCHAR(500) NOT NULL, {Configuration.PayloadColumnName} {storedAs} NOT NULL, - {Configuration.TagsColumnName} VARCHAR(100) NULL, + {Configuration.TagsColumnName} VARCHAR(100)[] NULL, {Configuration.SerializerIdColumnName} INTEGER NULL, CONSTRAINT {Configuration.JournalEventsTableName}_uq UNIQUE ({Configuration.PersistenceIdColumnName}, {Configuration.SequenceNrColumnName}) ); + + CREATE INDEX IF NOT EXISTS idx_{Configuration.FullJournalTableName.Replace('.', '_')}_{Configuration.TagsColumnName}_gin + ON {Configuration.FullJournalTableName} USING gin ({Configuration.TagsColumnName}); "; CreateMetaTableSql = $@" @@ -55,6 +69,19 @@ public PostgreSqlQueryExecutor(PostgreSqlQueryConfiguration configuration, Akka. CONSTRAINT {Configuration.MetaTableName}_pk PRIMARY KEY ({Configuration.PersistenceIdColumnName}, {Configuration.SequenceNrColumnName}) );"; + HighestTagOrderingSql = + $@" + SELECT MAX(e.{Configuration.OrderingColumnName}) as Ordering + FROM {Configuration.FullJournalTableName} e + WHERE e.{Configuration.OrderingColumnName} > @Ordering AND e.{Configuration.TagsColumnName} @> @Tag"; + + ByTagSql = + $@" + SELECT {allEventColumnNames}, e.{Configuration.OrderingColumnName} as Ordering + FROM {Configuration.FullJournalTableName} e + WHERE e.{Configuration.OrderingColumnName} > @Ordering AND e.{Configuration.TagsColumnName} @> @Tag + ORDER BY {Configuration.OrderingColumnName} ASC"; + switch (_configuration.StoredAs) { case StoredAsType.ByteA: @@ -75,7 +102,7 @@ public PostgreSqlQueryExecutor(PostgreSqlQueryConfiguration configuration, Akka. var deserializer = Serialization.FindSerializerForType(type, Configuration.DefaultSerializer); return deserializer.FromBinary((byte[])serialized, type); } - }; + }; break; case StoredAsType.JsonB: _serialize = e => new SerializationResult(NpgsqlDbType.Jsonb, JsonConvert.SerializeObject(e.Payload, _configuration.JsonSerializerSettings), null); @@ -93,6 +120,8 @@ public PostgreSqlQueryExecutor(PostgreSqlQueryConfiguration configuration, Akka. protected override DbCommand CreateCommand(DbConnection connection) => ((NpgsqlConnection)connection).CreateCommand(); protected override string CreateEventsJournalSql { get; } protected override string CreateMetaTableSql { get; } + protected override string HighestTagOrderingSql { get; } + protected override string ByTagSql { get; } protected override void WriteEvent(DbCommand command, IPersistentRepresentation e, IImmutableSet tags) { @@ -125,17 +154,61 @@ protected override void WriteEvent(DbCommand command, IPersistentRepresentation command.Parameters.Add(new NpgsqlParameter("@Payload", serializationResult.DbType) { Value = serializationResult.Payload }); - if (tags.Count != 0) + command.Parameters.Add(tags.Count != 0 + ? new NpgsqlParameter("@Tag", NpgsqlDbType.Array | NpgsqlDbType.Varchar) {Value = tags.ToArray()} + : new NpgsqlParameter("@Tag", NpgsqlDbType.Array | NpgsqlDbType.Varchar) {Value = DBNull.Value}); + } + + /// + /// TBD + /// + /// TBD + /// TBD + /// TBD + /// TBD + /// TBD + /// TBD + /// TBD + /// TBD + public override async Task SelectByTagAsync(DbConnection connection, CancellationToken cancellationToken, string tag, long fromOffset, long toOffset, long max, + Action callback) + { + using (var command = GetCommand(connection, ByTagSql)) { - var tagBuilder = new StringBuilder(";", tags.Sum(x => x.Length) + tags.Count + 1); - foreach (var tag in tags) + var take = Math.Min(toOffset - fromOffset, max); + command.Parameters.Add(new NpgsqlParameter("@Tag", NpgsqlDbType.Array | NpgsqlDbType.Varchar) { Value = new[] { tag }}); + AddParameter(command, "@Ordering", DbType.Int64, fromOffset); + AddParameter(command, "@Take", DbType.Int64, take); + + CommandBehavior commandBehavior; + + if (Configuration.UseSequentialAccess) + { + commandBehavior = CommandBehavior.SequentialAccess; + } + else + { + commandBehavior = CommandBehavior.Default; + } + + using (var reader = await command.ExecuteReaderAsync(commandBehavior, cancellationToken)) { - tagBuilder.Append(tag).Append(';'); + while (await reader.ReadAsync(cancellationToken)) + { + var persistent = ReadEvent(reader); + var ordering = reader.GetInt64(OrderingIndex); + callback(new ReplayedTaggedMessage(persistent, tag, ordering)); + } } + } - AddParameter(command, "@Tag", DbType.String, tagBuilder.ToString()); + using (var command = GetCommand(connection, HighestTagOrderingSql)) + { + command.Parameters.Add(new NpgsqlParameter("@Tag", NpgsqlDbType.Array | NpgsqlDbType.Varchar) { Value = new[] { tag } }); + AddParameter(command, "@Ordering", DbType.Int64, fromOffset); + var maxOrdering = (await command.ExecuteScalarAsync(cancellationToken)) as long? ?? 0L; + return maxOrdering; } - else AddParameter(command, "@Tag", DbType.String, DBNull.Value); } private static string QualifiedName(IPersistentRepresentation e) @@ -169,7 +242,7 @@ protected override IPersistentRepresentation ReadEvent(DbDataReader reader) return new Persistent(deserialized, sequenceNr, persistenceId, manifest, isDeleted, ActorRefs.NoSender, null); } } - + public class PostgreSqlQueryConfiguration : QueryConfiguration { public readonly StoredAsType StoredAs; @@ -191,10 +264,10 @@ public PostgreSqlQueryConfiguration( TimeSpan timeout, StoredAsType storedAs, string defaultSerializer, - JsonSerializerSettings jsonSerializerSettings = null, + JsonSerializerSettings jsonSerializerSettings = null, bool useSequentialAccess = true) : base(schemaName, journalEventsTableName, metaTableName, persistenceIdColumnName, sequenceNrColumnName, - payloadColumnName, manifestColumnName, timestampColumnName, isDeletedColumnName, tagsColumnName, orderingColumn, + payloadColumnName, manifestColumnName, timestampColumnName, isDeletedColumnName, tagsColumnName, orderingColumn, serializerIdColumnName, timeout, defaultSerializer, useSequentialAccess) { StoredAs = storedAs; From a496338034d57b72e0ecbe62074c645cadb2d2bd Mon Sep 17 00:00:00 2001 From: David Cumps Date: Mon, 2 Mar 2020 19:38:13 +0100 Subject: [PATCH 2/2] follow coding style for usings --- .../Journal/PostgreSqlQueryExecutor.cs | 5 ++--- 1 file changed, 2 insertions(+), 3 deletions(-) diff --git a/src/Akka.Persistence.PostgreSql/Journal/PostgreSqlQueryExecutor.cs b/src/Akka.Persistence.PostgreSql/Journal/PostgreSqlQueryExecutor.cs index f3c7844..638a23a 100755 --- a/src/Akka.Persistence.PostgreSql/Journal/PostgreSqlQueryExecutor.cs +++ b/src/Akka.Persistence.PostgreSql/Journal/PostgreSqlQueryExecutor.cs @@ -17,12 +17,11 @@ using System.Data; using System.Data.Common; using System.Linq; +using System.Threading; +using System.Threading.Tasks; namespace Akka.Persistence.PostgreSql.Journal { - using System.Threading; - using System.Threading.Tasks; - public class PostgreSqlQueryExecutor : AbstractQueryExecutor { private readonly PostgreSqlQueryConfiguration _configuration;