Tag table with backwards compatibility #68

to11mtm · 2022-03-19T21:26:44Z

Fixes #67

(Alternative solution to akkadotnet/akka.net#5296)

Changes

This is an implementation of Tag tables with a form of backwards compatibility.

Added TagWriteMode enum, with 3 modes of operation.
- applied logic to ByteArrayJournalSerializer so that tags may be applied based on setting
- Added logic to BaseByteArrayJournalDao to switch behavior based on mode of operation
Added logic to BaseByteArrayJournalDao to interleave batch-untagged and per-row-tagged writes when writing to tag table
Added TagReadMode enum, with 3 modes of operation.
- Added logic to BaseByteReadArrayJournalDao to respect enumeration
  - Queries running off tags will check tag table and/or main table based on setting
  - Filling of tags for stream pipeline will respect setting

Todo:

Some cleanup in the Read Journal and elsewhere
Config bits (defaults, reference, table/column names)
Create table bits
Tests around new functionality
- Ideally, we should have tests around migration scenarios.
Update deletion logic to delete tags in tag table
- Note: a combination of 'delete-compatibility-mode' and tag-tables may make deletions very painful, since you'd be locking across 3 tables.
Update 'update' functionality so that it will update tag table if needed.
- the update functionality is a non-standard addition present from persistence-jdbc.

Not in scope of this PR:

Migration routines/scripts. I think a helper application similar to what has been proposed for indexes would be a good idea, but this is likely a big enough PR as-is.

Checklist

For significant changes, please ensure that the following have been completed (delete if not relevant):

This change follows the Akka.NET API Compatibility Guidelines.
This change follows the Akka.NET Wire Compatibility Guidelines.
I have reviewed my own pull request.
Design discussion issue #
Changes in public API reviewed, if any.
I have added website documentation for this feature.

Latest `dev` Benchmarks

Include data from the relevant benchmark prior to this change here.

This PR's Benchmarks

Include data from after this change here.

…written even if empty

…tes with tags at expense of size and read speed. Cleaned up ReadJournal Queries

Aaronontheweb · 2022-06-14T21:23:38Z

Going to review this, as I am very interested in having this functionality for a project.

Aaronontheweb

Have a partial review for you @to11mtm - have about half the files to look at still.

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

Aaronontheweb · 2022-06-16T18:45:46Z

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

@@ -3,6 +3,19 @@

 namespace Akka.Persistence.Sql.Linq2Db.Config
 {
+    [Flags]
+    public enum TagWriteMode


Got it - this is how we can support data written in previous journal implementations.

Aaronontheweb · 2022-06-16T18:46:15Z

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

+        public TagWriteMode TagWriteMode { get; }
+        public TagTableMode TagTableMode { get; }
+        public string? TagTableName { get; }
+        public bool UseEventManifestColumn { get; }


These are all of the aforementioned compat-flags?

Aaronontheweb · 2022-06-16T18:47:25Z

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

+            }
+            else
+            {
+                res = TagWriteMode.CommaSeparatedArray;


Got it, this is probably how most popular platforms have handled this.

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

Aaronontheweb · 2022-06-16T18:59:28Z

src/Akka.Persistence.Sql.Linq2Db/Db/AkkaPersistenceDataConnectionFactory.cs

@@ -135,7 +135,13 @@ public AkkaPersistenceDataConnectionFactory(IProviderConfig<SnapshotTableConfigu
                .HasColumnName(tableConfig.ColumnNames.SequenceNumber)
                .Member(r => r.Timestamp)
                .HasColumnName(tableConfig.ColumnNames.Created);
-
+            //We can skip writing tags the old way by ignoring the column in mapping.
+            journalRowBuilder.Member(r => r.tagArr).IsNotColumn();


src/Akka.Persistence.Sql.Linq2Db/Db/AkkaPersistenceDataConnectionFactory.cs

Aaronontheweb · 2022-06-16T19:26:26Z

src/Akka.Persistence.Sql.Linq2Db/Journal/DAO/BaseByteArrayJournalDao.cs

@@ -23,6 +24,37 @@

 namespace Akka.Persistence.Sql.Linq2Db.Journal.DAO
 {
+    public class SequentialUUIDGenerator


Does akkadotnet/akka.net#5995 help with this?

Aaronontheweb · 2022-06-16T19:26:57Z

src/Akka.Persistence.Sql.Linq2Db/Journal/DAO/BaseByteArrayJournalDao.cs

+                var dbid = await dc.InsertWithInt64IdentityAsync(journalRow);
+                foreach (var s1 in journalRow.tagArr)
+                {
+                    tagsToInsert.Add(new JournalTagRow(){JournalOrderingId = dbid, TagValue = s1});


@Aaronontheweb TBH we could really use more tests around things, but I have never used tags so I have a hard time coming up with scenarios.

ismaelhamed · 2022-06-18T09:07:49Z

Is this a different approach than #467?

to11mtm · 2022-06-21T16:57:13Z

Is this a different approach than #467?

Yes and not quite:

Yes:

If you are running based off Ordering, it is close to the same approach as 467, with the added kindness of having migration modes. (The migration modes add a good bit of complexity here and are where much of the deviation occurs.)

Not Quite:

We aren't subclassing the way JDBC does to handle the different behavior modes. Given we are trying to provide migration modes, it would have meant 3 classes and a more complex migration process.
We are adding the option of using some form of SequentialUUID for joining to the tag table rather than orderingID.
- This is a mode intended for scenarios where high write throughput with tags is desired, at the expense of additional storage and possibly (likely) slower reads.
- This may not be a good idea, or may require additional optimization
We may or may not be doing the Inserts the same way SlickDB does. Based on the initial PRs for Tag tables in persistence-jdbc, I'm -guessing- their approach was similar to ours (where you have to insert each row one at a time to get IDs). Assumed this is still true based on them having split logic similar to this OR, where if there are no tags they try to do a simpler write.
- To solve this, Linq2Db will need to have multirow insert-with-return support, Or BulkCopy(Async) Returning Inserted IDs linq2db/linq2db#2960 needs to be resolved.

to11mtm · 2022-04-17T20:33:24Z

src/Akka.Persistence.Sql.Linq2Db/Db/AkkaPersistenceDataConnectionFactory.cs

+            if (config.TableConfig.UseEventManifestColumn)
+            {
+                journalRowBuilder.Member(r => r.eventManifest)
+                    .IsColumn().HasLength(64);
+            }
+            else
+            {
+                journalRowBuilder.Member(r => r.eventManifest)
+                    .IsNotColumn();   
+            }


Basically, this switch should control whether the 'fixed' behavior for eventManifest is used.

If it's not mapped, the property will come back null and the other logic works as it should, and is ignored on inserts.

to11mtm · 2022-04-17T20:37:37Z

src/Akka.Persistence.Sql.Linq2Db/Journal/DAO/BaseByteArrayJournalDao.cs

@@ -23,6 +24,37 @@

 namespace Akka.Persistence.Sql.Linq2Db.Journal.DAO
 {
+    public class SequentialUUIDGenerator


This algo is borrowed from EF Core, and should be attributed such and re-commented for our use.

The expectation is that we should be able to generate Guid-esque structures with some 'predictability' in a range of output.

This function's use may require fine tuning to produce optimal results.

Can we use akkadotnet/akka.net#5995 for this instead? I'd prefer to keep 1 implementation floating around, but if that approach doesn't make sense then no worries.

to11mtm · 2022-04-17T20:45:32Z

src/Akka.Persistence.Sql.Linq2Db/Query/Dao/BaseByteReadArrayJournalDAO.cs

+                .Via(deserializeFlow);
+        }
+
+        private Expression<Func<JournalRow, bool>> eventsByTagMigrationPredicate(DataConnection conn, string tagVal)


This had to be expanded into it's own method because of the compiler version used in build.

to11mtm · 2022-04-17T20:46:35Z

src/Akka.Persistence.Sql.Linq2Db/Query/Dao/BaseByteReadArrayJournalDAO.cs


+        private Expression<Func<JournalRow, JournalTagRow, bool>> EventsByTagOnlyJoinPredicate


This had to be expanded as such due to build server compiler not being able to infer expressions being returned from a ternary.

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

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

to11mtm · 2022-07-09T20:19:43Z

src/Akka.Persistence.Sql.Linq2Db/Journal/DAO/BaseByteArrayJournalDao.cs

@@ -23,6 +24,37 @@

 namespace Akka.Persistence.Sql.Linq2Db.Journal.DAO
 {
+    public class SequentialUUIDGenerator


to11mtm · 2022-07-09T20:20:47Z

src/Akka.Persistence.Sql.Linq2Db/Journal/DAO/BaseByteArrayJournalDao.cs

+                var dbid = await dc.InsertWithInt64IdentityAsync(journalRow);
+                foreach (var s1 in journalRow.tagArr)
+                {
+                    tagsToInsert.Add(new JournalTagRow(){JournalOrderingId = dbid, TagValue = s1});


@Aaronontheweb TBH we could really use more tests around things, but I have never used tags so I have a hard time coming up with scenarios.

Aaronontheweb · 2022-07-16T16:40:33Z

@to11mtm I'll review this again this week. Had it on my daily to-do list for a while but would really like to see it merged and Akka.Persistence.Linq2Db become the new standard ;)

Aaronontheweb

LGTM - however, for testing this prior to going into production I'd like to propose the following:

Create 3 databases: Akka.Persistence.SqlServer, Postgres, and MySQL with pre-populated data using tags - all created from the same application that uses the same persistent queries. Save those DB files and commit them into a repo somewhere.
Build a docker image that mounts and restores those databases.
Run an application that uses Akka.Persistence.Linq2Db's implementation of the journal and see if we can re-run the existing queries and produce the same results against all 3.
Re-run a version of this using custom object serialization, just to ensure that is also supported in the migration.

That would pass the snuff test for me to a degree where I'd feel comfortable deprecating Akka.Persistence.Sql.Common and its subsequent implementations. Our team at Petabridge can help with this. What do you think @to11mtm @ismaelhamed ?

Aaronontheweb · 2022-07-18T13:42:42Z

Build is failing due to NuGet.org errors at the moment.

Aaronontheweb · 2022-07-19T12:21:38Z

@to11mtm what do you think of my review comment above? Should we mark this as "ready" and merge it in, with a plan to complete those above steps with help from our team?

to11mtm · 2022-07-19T16:35:11Z

LGTM - however, for testing this prior to going into production I'd like to propose the following:

It's not done. That TODO list at the top for starters.

1. Create 3 databases: Akka.Persistence.SqlServer, Postgres, and MySQL with pre-populated data using tags - all created from the same application that uses the same persistent queries. Save those DB files and commit them into a repo somewhere.

You can't do MySQL until #3 is completed.

2. Build a docker image that mounts and restores those databases.

3. Run an application that uses Akka.Persistence.Linq2Db's implementation of the journal and see if we can re-run the existing queries and produce the same results against all 3.

4. Re-run a version of this using custom object serialization, just to ensure that is also supported in the migration.

That would pass the snuff test for me to a degree where I'd feel comfortable deprecating Akka.Persistence.Sql.Common and its subsequent implementations.

The other filed issues should be considered before deprecating Sql.Common, timestamp generator is one that comes to mind.

Our team at Petabridge can help with this. What do you think @to11mtm @ismaelhamed ?

As I've mentioned in other conversations this is not a task I've had the bandwidth to take over the finish line, both in scope of the TODOs (bandwidth) as well as more full testing (I'm not familiar enough with query to come up with good tests). I can answer questions and advise but as it stands the bus factor on this is too low.

As another side note, originally we hoped to deprecate sql.common earlier. As it stands, we risk a large influx of support issues and will risk more noise between compatibility issues of switching from sql.common vs whatever set of compatibility flags they choose (i.e. eventmanifest but no tag table, compatibility modes on different DBs... delete surprises...) something to consider.

tl;dr- TODOs need to be done before this gets merged, until then the tests you describe give a false sense this is done.

# Conflicts: # src/Akka.Persistence.Sql.Linq2Db/Config/BaseByteArrayJournalDaoConfig.cs # src/Akka.Persistence.Sql.Linq2Db/Config/JournalConfig.cs # src/Akka.Persistence.Sql.Linq2Db/Config/JournalTableConfig.cs # src/Akka.Persistence.Sql.Linq2Db/Config/ReadJournalPluginConfig.cs # src/Akka.Persistence.Sql.Linq2Db/Db/AkkaPersistenceDataConnectionFactory.cs # src/Akka.Persistence.Sql.Linq2Db/Journal/DAO/BaseByteArrayJournalDao.cs # src/Akka.Persistence.Sql.Linq2Db/Journal/DAO/BaseJournalDaoWithReadMessages.cs # src/Akka.Persistence.Sql.Linq2Db/Journal/DAO/ByteArrayJournalSerializer.cs # src/Akka.Persistence.Sql.Linq2Db/Query/Dao/BaseByteReadArrayJournalDAO.cs # src/Akka.Persistence.Sql.Linq2Db/Query/JournalSequenceActor.cs # src/Akka.Persistence.Sql.Linq2Db/Serialization/PersistentReprSerializer.cs

# Conflicts: # Akka.Persistence.Linq2Db.sln

# Conflicts: # src/Akka.Persistence.Sql.Linq2Db.Tests/Settings/ReadJournalConfigSpec.cs # src/Akka.Persistence.Sql.Linq2Db/Config/ReadJournalConfig.cs # src/Akka.Persistence.Sql.Linq2Db/Query/Dao/BaseByteReadArrayJournalDAO.cs # src/Akka.Persistence.Sql.Linq2Db/persistence.conf

Aaronontheweb · 2022-12-01T16:35:43Z

@Arkatufus you should move your commits onto its own PR and try to leave @to11mtm 's work here intact, for the sake of making it easy to compare the difference between the two separately.

# Conflicts: # src/Akka.Persistence.Linq2Db.Compatibility.Tests/SqlCommonJournalCompatibilitySpec.cs # src/Akka.Persistence.Linq2Db.Compatibility.Tests/SqlCommonSnapshotCompatibilitySpec.cs

# Conflicts: # src/Akka.Persistence.Linq2Db.Compatibility.DockerTests/SqlServer/SqlServerCompatibilitySpecConfig.cs # src/Akka.Persistence.Sql.Linq2Db/Config/JournalTableConfig.cs # src/Akka.Persistence.Sql.Linq2Db/persistence.conf

Aaronontheweb

Approving and merging this because it's a blocker and I don't want to review another 90+ file PR again, but there are some issues that need to be addressed in the comments. Let's deal with those in separate PRs.

Aaronontheweb · 2023-02-03T16:05:18Z

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

+    <ItemGroup>
+      <PackageReference Include="Akka.Streams" Version="1.4.21" />
+      <PackageReference Include="linq2db" Version="3.4.1" />
+      <PackageReference Include="Reactive.Streams" Version="1.0.2" />


No need for an explicit Reactive.Streams reference here

Aaronontheweb · 2023-02-03T16:07:04Z

....Persistence.Linq2Db.Compatibility.Tests/Akka.Persistence.Linq2Db.Compatibility.Tests.csproj

@@ -9,6 +9,7 @@
    </PropertyGroup>

    <ItemGroup>
+        <PackageReference Include="Akka.TestKit" Version="$(AkkaVersion)" />
        <PackageReference Include="Akka" Version="$(AkkaVersion)" />


Can remove explicit Akka reference if we're referencing the TestKit

Aaronontheweb · 2023-02-03T16:12:04Z

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

@@ -9,6 +9,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);


Aaronontheweb · 2023-02-03T16:12:43Z

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

@@ -0,0 +1,41 @@
+// -----------------------------------------------------------------------
+//  <copyright file="TagTableConfig.cs" company="Akka.NET Project">
+//      Copyright (C) 2009-2023 Lightbend Inc. <http://www.lightbend.com>


Can get rid of Lightbend attribution here.

Aaronontheweb · 2023-02-03T16:12:52Z

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

@@ -0,0 +1,43 @@
+// -----------------------------------------------------------------------
+//  <copyright file="TagTableColumnNames.cs" company="Akka.NET Project">
+//      Copyright (C) 2009-2023 Lightbend Inc. <http://www.lightbend.com>


Aaronontheweb · 2023-02-03T16:14:01Z

src/Akka.Persistence.Sql.Linq2Db/Db/AkkaPersistenceDataConnectionFactory.cs

@@ -140,6 +140,13 @@ public AkkaPersistenceDataConnectionFactory(IProviderConfig<SnapshotTableConfigu
                .Member(r => r.Timestamp)
                .HasColumnName(columnNames.Created);

+            //We can skip writing tags the old way by ignoring the column in mapping.
+            journalRowBuilder.Member(r => r.TagArr).IsNotColumn();
+            if (tableConfig.TagWriteMode == TagWriteMode.TagTable)


Based on @to11mtm 's suggestion, we're going to need to be able to support multiple tag write modes. Can we address this in a subsequent PR?

Aaronontheweb · 2023-02-03T16:14:35Z

src/Akka.Persistence.Sql.Linq2Db/Journal/DAO/BaseByteArrayJournalDao.cs

 using LinqToDB.Data;
 using static LanguageExt.Prelude;

 namespace Akka.Persistence.Sql.Linq2Db.Journal.Dao
 {
+    public class SequentialUuidGenerator


Can make this static

Aaronontheweb · 2023-02-03T16:15:08Z

src/Akka.Persistence.Sql.Linq2Db/Journal/DAO/ByteArrayJournalSerializer.cs

@@ -18,13 +19,15 @@ public sealed class ByteArrayJournalSerializer : FlowPersistentReprSerializer<Jo
        private readonly string _separator;
        private readonly IProviderConfig<JournalTableConfig> _journalConfig;
        private readonly string[] _separatorArray;
+        private readonly TagWriteMode _tagWriteMode;


Consideration for multiple write modes here as well

Aaronontheweb · 2023-02-03T16:16:50Z

src/Akka.Persistence.Sql.Linq2Db/Journal/Types/JournalRow.cs

+
+        public string TagValue { get; set; }
+
+        public Guid WriteUuid { get; set; }


No need to include WriteUuid on the column - OrderingId should be good enough if it's unique.

Aaronontheweb · 2023-02-03T16:18:44Z

src/Akka.Persistence.Sql.Linq2Db/persistence.conf

+          columns {
+            ordering-id = ordering_id
+            tag-value = tag
+            writer-uuid = writer_uuid


Can remove this

to11mtm added 3 commits January 9, 2022 15:09

index creation app example

660f637

WIP Tag Table with backwards compatibility

fec6cf5

fix incomplete rename refactor, re-add case to guarantee manifest is …

8b65724

…written even if empty

to11mtm mentioned this pull request Mar 20, 2022

Feature Req: Create a Helper application to help users migrate away from journal_metadata #69

Open

to11mtm added 5 commits March 20, 2022 15:42

fix: code for journal queries was not always respecting isdeleted

adc8ce4

Added WIP Tag-join-via UUID Support for users who may want faster wri…

6936c1e

…tes with tags at expense of size and read speed. Cleaned up ReadJournal Queries

Try refactoring to make older compiler happy about expressions.

627d159

Try Unwinding ternarys to make compiler happy.

bc52579

WIP Allow proper handling of EventManifest

3778099

Aaronontheweb self-requested a review June 14, 2022 21:23

Aaronontheweb reviewed Jun 16, 2022

View reviewed changes

to11mtm commented Jul 16, 2022

View reviewed changes

Aaronontheweb added 2 commits July 18, 2022 08:37

Merge branch 'dev' into tag-table-with-backwards-compat

4c2dc03

Rename Class1.cs to JournalIndexHelper.cs

34570b9

Aaronontheweb approved these changes Jul 18, 2022

View reviewed changes

Arkatufus added 2 commits November 23, 2022 20:48

Move dangling projects into src folder

f5e59b3

Arkatufus mentioned this pull request Nov 23, 2022

index creation app example #55

Closed

6 tasks

Arkatufus added 5 commits November 23, 2022 21:46

Cleanup property names

56777fa

Merge branch 'dev' into tag-table-with-backwards-compat

f67450d

# Conflicts: # Akka.Persistence.Linq2Db.sln

Fix merge issues

d4e2984

Fix table column name compatibility problem

060da94

Fix logical delete compatibility problem

5a4595b

Arkatufus added 5 commits November 28, 2022 20:59

Merge branch 'dev' into tag-table-with-backwards-compat

67bd601

post-merge cleanup

095b0dd

Convert unit tests to use TestKit

969116b

Fix unit tests

bfe4f86

Merge branch 'dev' into tag-table-with-backwards-compat

3455e3c

# Conflicts: # src/Akka.Persistence.Linq2Db.Compatibility.Tests/SqlCommonJournalCompatibilitySpec.cs # src/Akka.Persistence.Linq2Db.Compatibility.Tests/SqlCommonSnapshotCompatibilitySpec.cs

Arkatufus marked this pull request as ready for review January 31, 2023 20:59

Arkatufus added 4 commits February 2, 2023 01:14

Add SqlServer SQL script migration spec

d7a283d

Add PostgreSql and MySql SQL script spec

b2a3329

Remove bit flag

c59c980

Aaronontheweb approved these changes Feb 3, 2023

View reviewed changes

Aaronontheweb merged commit a26036e into akkadotnet:dev Feb 3, 2023


		private Expression<Func<JournalRow, JournalTagRow, bool>> EventsByTagOnlyJoinPredicate


		public string TagValue { get; set; }

		public Guid WriteUuid { get; set; }

Tag table with backwards compatibility #68

Tag table with backwards compatibility #68

Conversation

to11mtm commented Mar 19, 2022 • edited

Changes

Todo:

Not in scope of this PR:

Checklist

Latest dev Benchmarks

This PR's Benchmarks

Aaronontheweb commented Jun 14, 2022

Aaronontheweb left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

ismaelhamed commented Jun 18, 2022

to11mtm commented Jun 21, 2022

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Aaronontheweb commented Jul 16, 2022

Aaronontheweb left a comment

Choose a reason for hiding this comment

Aaronontheweb commented Jul 18, 2022

Aaronontheweb commented Jul 19, 2022

to11mtm commented Jul 19, 2022

Aaronontheweb commented Dec 1, 2022

Aaronontheweb left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

to11mtm commented Mar 19, 2022 •

edited

Latest `dev` Benchmarks