Replay optimized postgres persistor fixes

lib/sequent/core/persistors/replay_optimized_postgres_persistor.rb

@@ -72,39 +72,34 @@ def set_values(values)
         class Index
           def initialize(indexed_columns)
             @indexed_columns = Hash.new do |hash, record_class|
-              hash[record_class] = if record_class.column_names.include? 'aggregate_id'
-                                     ['aggregate_id']
-                                   else
-                                     []
-                                   end
+              hash[record_class] = default_indexes(record_class)
             end
 
-            @indexed_columns = @indexed_columns.merge(
-              indexed_columns.reduce({}) do |memo, (key, ics)|
-                memo.merge({key => ics.map { |c| c.map(&:to_s) }})
-              end,
-            )
+            indexed_columns.each do |record_class, indexes|
+              normalized = indexes.map { |index| index.map(&:to_s).sort }
+              @indexed_columns[record_class] = (normalized + default_indexes(record_class)).uniq
+            end
 
             @index = {}
-            @reverse_index = {}
+            @reverse_index = {}.compare_by_identity
           end
 
           def add(record_class, record)
             return unless indexed?(record_class)
 
             get_keys(record_class, record).each do |key|
-              @index[key.hash] = [] unless @index.key? key.hash
-              @index[key.hash] << record
+              @index[key] = [] unless @index.key? key
+              @index[key] << record
 
-              @reverse_index[record.object_id.hash] = [] unless @reverse_index.key? record.object_id.hash
-              @reverse_index[record.object_id.hash] << key.hash
+              @reverse_index[record] = [] unless @reverse_index.key? record
+              @reverse_index[record] << key
             end
           end
 
           def remove(record_class, record)
             return unless indexed?(record_class)
 
-            keys = @reverse_index.delete(record.object_id.hash) { [] }
+            keys = @reverse_index.delete(record) { [] }
 
             return unless keys.any?
 
@@ -120,12 +115,12 @@ def update(record_class, record)
           end
 
           def find(record_class, where_clause)
-            key = [record_class.name]
-            get_index(record_class, where_clause).each do |field|
-              key << field
-              key << where_clause.stringify_keys[field]
-            end
-            @index[key.hash] || []
+            index = get_index(record_class, where_clause)
+            return nil unless index
+
+            normalized_where_clause = where_clause.stringify_keys
+            key = [record_class.name, index] + index.map { |field| normalized_where_clause[field] }
+            @index[key] || []
           end
 
           def clear
@@ -134,29 +129,34 @@ def clear
           end
 
           def use_index?(record_class, where_clause)
-            @indexed_columns.key?(record_class) && get_index(record_class, where_clause).present?
+            indexed?(record_class) && get_index(record_class, where_clause).present?
           end
 
           private
 
           def indexed?(record_class)
-            @indexed_columns.key?(record_class)
+            # Do not use `key?` here or similar, since the
+            # `@indexed_columns#default_proc` automatically adds new
+            # indexes as required.
+            @indexed_columns[record_class].present?
           end
 
           def get_keys(record_class, record)
             @indexed_columns[record_class].map do |index|
-              arr = [record_class.name]
-              index.each do |key|
-                arr << key
-                arr << record[key]
-              end
-              arr
+              [record_class.name, index] + index.map { |field| record[field] }
             end
           end
 
           def get_index(record_class, where_clause)
-            @indexed_columns[record_class].find do |indexed_where|
-              where_clause.keys.size == indexed_where.size && (where_clause.keys.map(&:to_s) - indexed_where).empty?
+            where_clause_keys = where_clause.keys.map(&:to_s).sort
+            @indexed_columns[record_class].find { |index| index == where_clause_keys }
+          end
+
+          def default_indexes(record_class)
+            if record_class.column_names.include? 'aggregate_id'
+              [['aggregate_id']]
+            else
+              []
             end
           end
         end
@@ -287,22 +287,18 @@ def do_with_record(record_class, where_clause)
         end
 
         def find_records(record_class, where_clause)
-          if @record_index.use_index?(record_class, where_clause)
-            @record_index.find(record_class, where_clause)
-          else
-            @record_store[record_class].select do |record|
-              where_clause.all? do |k, v|
-                expected_value = v.is_a?(Symbol) ? v.to_s : v
-                actual_value = record[k.to_sym]
-                actual_value = actual_value.to_s if actual_value.is_a? Symbol
-                if expected_value.is_a?(Array)
-                  expected_value.include?(actual_value)
-                else
-                  actual_value == expected_value
-                end
+          (@record_index.find(record_class, where_clause) || @record_store[record_class].select do |record|
+            where_clause.all? do |k, v|
+              expected_value = v.is_a?(Symbol) ? v.to_s : v
+              actual_value = record[k.to_sym]
+              actual_value = actual_value.to_s if actual_value.is_a? Symbol
+              if expected_value.is_a?(Array)
+                expected_value.include?(actual_value)
+              else
+                actual_value == expected_value
               end
             end
-          end.dup
+          end).dup
         end
 
         def last_record(record_class, where_clause)

spec/lib/sequent/core/persistors/replay_optimized_postgres_persistor_spec.rb

@@ -12,6 +12,13 @@ def initialize
   end
 end
 
+def measure_elapsed_time(&block)
+  starting = Process.clock_gettime(Process::CLOCK_MONOTONIC)
+  yield block
+  ending = Process.clock_gettime(Process::CLOCK_MONOTONIC)
+  ending - starting
+end
+
 describe Sequent::Core::Persistors::ReplayOptimizedPostgresPersistor do
   let(:persistor) { Sequent::Core::Persistors::ReplayOptimizedPostgresPersistor.new }
   let(:record_class) { Sequent::Core::EventRecord }
@@ -378,17 +385,64 @@ class ReplayOptimizedPostgresTest < Sequent::ApplicationRecord; end
     end
   end
 
-  describe Sequent::Core::Persistors::ReplayOptimizedPostgresPersistor::Index do
-    describe '#use_index?' do
-      let(:indices) { [] }
-      let(:index) do
-        Sequent::Core::Persistors::ReplayOptimizedPostgresPersistor::Index.new(
-          {
-            Sequent::Core::EventRecord => indices,
-          },
+  context 'with thousands of records' do
+    COUNT = 1000
+    ITERATIONS = 10
+    MAX_TIME_S = 1
+
+    let(:persistor) do
+      Sequent::Core::Persistors::ReplayOptimizedPostgresPersistor.new(
+        50,
+        {
+          Sequent::Core::EventRecord => [%i[id command_record_id], %i[id sequence_number]],
+        },
+      )
+    end
+    let(:aggregate_ids) { (0...COUNT).map { Sequent.new_uuid } }
+
+    before do
+      aggregate_ids.each_with_index do |aggregate_id, i|
+        persistor.create_record(
+          Sequent::Core::EventRecord,
+          {id: i, aggregate_id: aggregate_id, command_record_id: i * 7},
         )
       end
+    end
 
+    it 'performs well using an aggregate_id lookup' do
+      elapsed = measure_elapsed_time do
+        ITERATIONS.times do
+          aggregate_ids.each do |aggregate_id|
+            expect(persistor.get_record(Sequent::Core::EventRecord, {aggregate_id: aggregate_id})).to be_present
+          end
+        end
+      end
+      expect(elapsed).to be <= MAX_TIME_S
+    end
+
+    it 'performs well using a multi-index lookup' do
+      elapsed = measure_elapsed_time do
+        ITERATIONS.times do
+          (0...COUNT).each do |i|
+            expect(persistor.get_record(Sequent::Core::EventRecord, {id: i, command_record_id: i * 7})).to be_present
+          end
+        end
+      end
+      expect(elapsed).to be <= MAX_TIME_S
+    end
+  end
+
+  describe Sequent::Core::Persistors::ReplayOptimizedPostgresPersistor::Index do
+    let(:indices) { [] }
+    let(:index) do
+      Sequent::Core::Persistors::ReplayOptimizedPostgresPersistor::Index.new(
+        {
+          Sequent::Core::EventRecord => indices,
+        },
+      )
+    end
+
+    describe '#use_index?' do
       context 'symbolized single indices' do
         let(:indices) { [[:id]] }
         it 'uses the index for strings and symbols where clause' do
@@ -427,6 +481,27 @@ class ReplayOptimizedPostgresTest < Sequent::ApplicationRecord; end
           expect(index.use_index?(Sequent::Core::EventRecord, {sequence_number: 1})).to be_falsey
           expect(index.use_index?(Sequent::Core::EventRecord, {id: 1, sequence_number: 1})).to be_falsey
         end
+
+        context 'duplicate indexes' do
+          let(:indices) { [%i[aggregate_id], %i[command_record_id id], %i[id command_record_id]] }
+          it 'are removed' do
+            expect(index.instance_variable_get(:@indexed_columns)[Sequent::Core::EventRecord])
+              .to match_array [['aggregate_id'], %w[command_record_id id]]
+          end
+        end
+      end
+
+      context 'default index when record class is specified' do
+        it 'adds a default index for aggregate_id' do
+          expect(index.use_index?(Sequent::Core::EventRecord, {aggregate_id: 1})).to be_truthy
+        end
+      end
+
+      context 'default index when record class is not specified' do
+        let(:index) { Sequent::Core::Persistors::ReplayOptimizedPostgresPersistor::Index.new({}) }
+        it 'adds a default index for aggregate_id' do
+          expect(index.use_index?(Sequent::Core::EventRecord, {aggregate_id: 1})).to be_truthy
+        end
       end
 
       context 'where clause order' do
@@ -437,5 +512,24 @@ class ReplayOptimizedPostgresTest < Sequent::ApplicationRecord; end
         end
       end
     end
+
+    context 'duplicate hash values' do
+      class BadHash < Struct.new(:value)
+        def hash
+          0
+        end
+      end
+
+      it 'should not match records even when hash collision occurs' do
+        one = persistor.create_record(Sequent::Core::EventRecord, aggregate_id: BadHash.new(1), sequence_number: 1)
+        two = persistor.create_record(Sequent::Core::EventRecord, aggregate_id: BadHash.new(2), sequence_number: 1)
+
+        index.add(Sequent::Core::EventRecord, one)
+        index.add(Sequent::Core::EventRecord, two)
+
+        expect(index.find(Sequent::Core::EventRecord, {aggregate_id: one.aggregate_id})).to match_array [one]
+        expect(index.find(Sequent::Core::EventRecord, {aggregate_id: two.aggregate_id})).to match_array [two]
+      end
+    end
   end
 end