Merge pull request #360 from SwissBorg/155-journal-metadata/upsert-mi…

…n-ordering Also upsert min_ordering on metadata trigger
SwissBorg · Sep 7, 2023 · 52b153a · 52b153a
2 parents 2bfafae + eccda7c
commit 52b153a
Show file tree

Hide file tree

Showing 6 changed files with 11 additions and 6 deletions.
diff --git a/core/src/test/resources/schema/postgres/nested-partitions-schema.sql b/core/src/test/resources/schema/postgres/nested-partitions-schema.sql
@@ -98,7 +98,8 @@ BEGIN
   ON CONFLICT (persistence_id) DO UPDATE
   SET
     max_sequence_number = GREATEST(public.journal_metadata.max_sequence_number, NEW.sequence_number),
-    max_ordering = GREATEST(public.journal_metadata.max_ordering, NEW.ordering);
+    max_ordering = GREATEST(public.journal_metadata.max_ordering, NEW.ordering),
+    min_ordering = LEAST(public.journal_metadata.min_ordering, NEW.ordering);
 
   RETURN NEW;
 END;

diff --git a/core/src/test/resources/schema/postgres/partitioned-schema.sql b/core/src/test/resources/schema/postgres/partitioned-schema.sql
@@ -99,7 +99,8 @@ BEGIN
   ON CONFLICT (persistence_id) DO UPDATE
   SET
     max_sequence_number = GREATEST(public.journal_metadata.max_sequence_number, NEW.sequence_number),
-    max_ordering = GREATEST(public.journal_metadata.max_ordering, NEW.ordering);
+    max_ordering = GREATEST(public.journal_metadata.max_ordering, NEW.ordering),
+    min_ordering = LEAST(public.journal_metadata.min_ordering, NEW.ordering);
 
   RETURN NEW;
 END;

diff --git a/core/src/test/resources/schema/postgres/plain-schema.sql b/core/src/test/resources/schema/postgres/plain-schema.sql
@@ -72,7 +72,8 @@ BEGIN
   ON CONFLICT (persistence_id) DO UPDATE
   SET
     max_sequence_number = GREATEST(public.journal_metadata.max_sequence_number, NEW.sequence_number),
-    max_ordering = GREATEST(public.journal_metadata.max_ordering, NEW.ordering);
+    max_ordering = GREATEST(public.journal_metadata.max_ordering, NEW.ordering),
+    min_ordering = LEAST(public.journal_metadata.min_ordering, NEW.ordering);
 
   RETURN NEW;
 END;

diff --git a/core/src/test/scala/akka/persistence/postgres/journal/PostgresJournalSpec.scala b/core/src/test/scala/akka/persistence/postgres/journal/PostgresJournalSpec.scala
@@ -122,7 +122,7 @@ abstract class PostgresJournalSpec(config: String, schemaType: SchemaType)
       minOrdering > 0 shouldBe true
     }
 
-    "upsert only max_sequence_number and max_ordering if metadata already exists" in {
+    "if metadata already exists only max_sequence_number and max_ordering should update and min_ordering should stay the same" in {
       // given
       val perId = "perId-meta-2"
       val sender = TestProbe()

diff --git a/migration/src/main/scala/akka/persistence/postgres/migration/journal/JournalSchema.scala b/migration/src/main/scala/akka/persistence/postgres/migration/journal/JournalSchema.scala
@@ -122,7 +122,8 @@ private[journal] trait JournalSchema {
               ON CONFLICT (#$persistenceId) DO UPDATE
               SET
                 #$maxSequenceNumber = GREATEST(#$fullTableName.#$maxSequenceNumber, NEW.#$sequenceNumber),
-                #$maxOrdering = GREATEST(#$fullTableName.#$maxOrdering, NEW.#$ordering);
+                #$maxOrdering = GREATEST(#$fullTableName.#$maxOrdering, NEW.#$ordering),
+                #$minOrdering = LEAST(#$fullTableName.#$minOrdering, NEW.#$ordering);
             
               RETURN NEW;
             END;

diff --git a/scripts/migration-0.6.0/2-create-function-update-journal-metadata.sql b/scripts/migration-0.6.0/2-create-function-update-journal-metadata.sql
@@ -30,7 +30,8 @@ BEGIN
   vals := '($1).' || j_persistence_id_column || ', ($1).' || j_sequence_number_column || ', ($1).' || j_ordering_column ||
           ', CASE WHEN ($1).' || j_sequence_number_column || ' = ' || first_sequence_number_value || ' THEN ($1).' || j_ordering_column || ' ELSE ' || unset_min_ordering_value || ' END';
   upds := jm_max_sequence_number_column || ' = GREATEST(' || jm_table || '.' || jm_max_sequence_number_column || ', ($1).' || j_sequence_number_column || '), ' ||
-          jm_max_ordering_column || ' = GREATEST(' || jm_table || '.' || jm_max_ordering_column || ', ($1).' || j_ordering_column || ')';
+          jm_max_ordering_column || ' = GREATEST(' || jm_table || '.' || jm_max_ordering_column || ', ($1).' || j_ordering_column || '), ' ||
+          jm_min_ordering_column || ' = LEAST(' || jm_table || '.' || jm_min_ordering_column || ', ($1).' || j_ordering_column || ')';
 
   sql := 'INSERT INTO ' || jm_table || ' (' || cols || ') VALUES (' || vals || ') ' ||
          'ON CONFLICT (' || jm_persistence_id_column || ') DO UPDATE SET ' || upds;