Hash by persistence_id only

Author: tiagomota

core/src/test/resources/schema/postgres/nested-partitions-schema.sql

@@ -77,8 +77,8 @@ CREATE TABLE public.journal_persistence_ids(
   max_sequence_number BIGINT NOT NULL,
   min_ordering BIGINT NOT NULL,
   max_ordering BIGINT NOT NULL,
-  PRIMARY KEY (id, persistence_id)
-) PARTITION BY HASH(id);
+  PRIMARY KEY (persistence_id)
+) PARTITION BY HASH(persistence_id);
 
 CREATE TABLE public.journal_persistence_ids_0 PARTITION OF public.journal_persistence_ids FOR VALUES WITH (MODULUS 2, REMAINDER 0);
 CREATE TABLE public.journal_persistence_ids_1 PARTITION OF public.journal_persistence_ids FOR VALUES WITH (MODULUS 2, REMAINDER 1);
@@ -89,7 +89,7 @@ DECLARE
 BEGIN
   INSERT INTO public.journal_persistence_ids (persistence_id, max_sequence_number, max_ordering, min_ordering)
   VALUES (NEW.persistence_id, NEW.sequence_number, NEW.ordering, NEW.ordering)
-  ON CONFLICT (id, persistence_id) DO UPDATE
+  ON CONFLICT (persistence_id) DO UPDATE
   SET
     max_sequence_number = NEW.sequence_number,
     max_ordering = NEW.ordering,

core/src/test/resources/schema/postgres/partitioned-schema.sql

@@ -78,8 +78,8 @@ CREATE TABLE public.journal_persistence_ids(
   max_sequence_number BIGINT NOT NULL,
   min_ordering BIGINT NOT NULL,
   max_ordering BIGINT NOT NULL,
-  PRIMARY KEY (id, persistence_id)
-) PARTITION BY HASH(id);
+  PRIMARY KEY (persistence_id)
+) PARTITION BY HASH(persistence_id);
 
 CREATE TABLE public.journal_persistence_ids_0 PARTITION OF public.journal_persistence_ids FOR VALUES WITH (MODULUS 2, REMAINDER 0);
 CREATE TABLE public.journal_persistence_ids_1 PARTITION OF public.journal_persistence_ids FOR VALUES WITH (MODULUS 2, REMAINDER 1);
@@ -90,7 +90,7 @@ DECLARE
 BEGIN
   INSERT INTO public.journal_persistence_ids (persistence_id, max_sequence_number, max_ordering, min_ordering)
   VALUES (NEW.persistence_id, NEW.sequence_number, NEW.ordering, NEW.ordering)
-  ON CONFLICT (id, persistence_id) DO UPDATE
+  ON CONFLICT (persistence_id) DO UPDATE
   SET
     max_sequence_number = NEW.sequence_number,
     max_ordering = NEW.ordering,

core/src/test/resources/schema/postgres/plain-schema.sql

@@ -51,8 +51,8 @@ CREATE TABLE public.journal_persistence_ids(
   max_sequence_number BIGINT NOT NULL,
   min_ordering BIGINT NOT NULL,
   max_ordering BIGINT NOT NULL,
-  PRIMARY KEY (id, persistence_id)
-) PARTITION BY HASH(id);
+  PRIMARY KEY (persistence_id)
+) PARTITION BY HASH(persistence_id);
 
 CREATE TABLE public.journal_persistence_ids_0 PARTITION OF public.journal_persistence_ids FOR VALUES WITH (MODULUS 2, REMAINDER 0);
 CREATE TABLE public.journal_persistence_ids_1 PARTITION OF public.journal_persistence_ids FOR VALUES WITH (MODULUS 2, REMAINDER 1);
@@ -63,7 +63,7 @@ DECLARE
 BEGIN
   INSERT INTO public.journal_persistence_ids (persistence_id, max_sequence_number, max_ordering, min_ordering)
   VALUES (NEW.persistence_id, NEW.sequence_number, NEW.ordering, NEW.ordering)
-  ON CONFLICT (id, persistence_id) DO UPDATE
+  ON CONFLICT (persistence_id) DO UPDATE
   SET
     max_sequence_number = NEW.sequence_number,
     max_ordering = NEW.ordering,

migration/src/main/scala/akka/persistence/postgres/migration/journal/JournalSchema.scala

@@ -37,8 +37,8 @@ private[journal] trait JournalSchema {
             #$maxSequenceNumber BIGINT       NOT NULL,
             #$maxOrdering       BIGINT       NOT NULL,
             #$minOrdering       BIGINT       NOT NULL,
-            PRIMARY KEY (#$id, #$persistenceId)
-          ) PARTITION BY HASH(#$id)"""
+            PRIMARY KEY (#$persistenceId)
+          ) PARTITION BY HASH(#$persistenceId)"""
       _ <-
         sqlu"""CREATE TABLE #${fullTableName}_0 PARTITION OF #$fullTableName FOR VALUES WITH (MODULUS 2, REMAINDER 0)"""
       _ <-
@@ -111,7 +111,7 @@ private[journal] trait JournalSchema {
             BEGIN
               INSERT INTO #$fullTableName (#$persistenceId, #$maxSequenceNumber, #$maxOrdering, #$minOrdering)
               VALUES (NEW.#$jPersistenceId, NEW.#$sequenceNumber, NEW.#$ordering, NEW.#$ordering)
-              ON CONFLICT (#$id, #$persistenceId) DO UPDATE
+              ON CONFLICT (#$persistenceId) DO UPDATE
               SET
                 #$maxSequenceNumber = NEW.#$sequenceNumber,
                 #$maxOrdering = NEW.#$ordering,

scripts/migration-0.6.0/1-create-journal-persistence-ids-table.sql

@@ -9,7 +9,7 @@ DECLARE
   jpi_max_sequence_number_column CONSTANT TEXT := 'max_sequence_number';
   jpi_max_ordering_column CONSTANT TEXT := 'max_ordering';
   jpi_min_ordering_column CONSTANT TEXT := 'min_ordering';
-  jpi_partitions_table_name_perfix CONSTANT TEXT := 'journal_persistence_ids_';
+  jpi_partitions_table_name_prefix CONSTANT TEXT := 'journal_persistence_ids_';
   jpi_partitions_number CONSTANT INTEGER := 10;
 
   -- variables
@@ -25,13 +25,13 @@ BEGIN
           jpi_max_sequence_number_column || ' BIGINT NOT NULL, ' ||
           jpi_max_ordering_column || ' BIGINT NOT NULL, ' ||
           jpi_min_ordering_column || ' BIGINT NOT NULL, ' ||
-          'PRIMARY KEY (' || jpi_id_column || ', ' || jpi_persistence_id_column || ')' ||
-        ') PARTITION BY HASH(' || jpi_id_column || ')';
+          'PRIMARY KEY (' || jpi_persistence_id_column || ')' ||
+        ') PARTITION BY HASH(' || jpi_persistence_id_column || ')';
 
   EXECUTE sql;
 
   FOR i IN 0..(jpi_partitions_number - 1) LOOP
-    EXECUTE 'CREATE TABLE IF NOT EXISTS ' || jpi_partitions_table_name_perfix || i ||
+    EXECUTE 'CREATE TABLE IF NOT EXISTS ' || jpi_partitions_table_name_prefix || i ||
             ' PARTITION OF ' || jpi_table ||
             ' FOR VALUES WITH (MODULUS ' || jpi_partitions_number || ', REMAINDER ' || i || ')';
   END LOOP;

scripts/migration-0.6.0/2-create-function-update-journal-persistence-ids.sql

@@ -9,7 +9,6 @@ DECLARE
   j_sequence_number_column CONSTANT TEXT := 'sequence_number';
   j_ordering_column CONSTANT TEXT := 'ordering';
   jpi_table_name CONSTANT TEXT := 'journal_persistence_ids';
-  jpi_id_column CONSTANT TEXT := 'id';
   jpi_persistence_id_column CONSTANT TEXT := 'persistence_id';
   jpi_max_sequence_number_column CONSTANT TEXT := 'max_sequence_number';
   jpi_max_ordering_column CONSTANT TEXT := 'max_ordering';
@@ -32,7 +31,7 @@ BEGIN
           jpi_min_ordering_column || ' = LEAST(' || jpi_table || '.' || jpi_min_ordering_column || ', ($1).' || j_ordering_column || ')';
 
   sql := 'INSERT INTO ' || jpi_table || ' (' || cols || ') VALUES (' || vals || ') ' ||
-         'ON CONFLICT (' || jpi_id_column || ', ' || jpi_persistence_id_column || ') DO UPDATE SET ' || upds;
+         'ON CONFLICT (' || jpi_persistence_id_column || ') DO UPDATE SET ' || upds;
 
   EXECUTE sql USING NEW;
   RETURN NEW;