Enumerate aggregates in aggregate_id order

Ordering by `[events_partition_key, aggregate_id]` is very inefficient
since the `aggregates` table is partitioned by
`aggregate_id`. Ordering by `aggregate_id` should allow fast index
scans. Retrieving the associated event streams will be slower though,
since they are no longer co-located by using the
`events_partition_key`.

Author: erikrozendaal

lib/sequent/core/event_store.rb

@@ -224,28 +224,24 @@ def load_events_since_marked_position(mark)
       # aggregate root type can be specified (as a string) to only yield aggregate ids of the indicated type.
       def event_streams_enumerator(aggregate_type: nil, group_size: 100)
         Enumerator.new do |yielder|
-          last_events_partition_key = ''
           last_aggregate_id = nil
           loop do
             aggregate_rows = ActiveRecord::Base.connection.exec_query(
-              'SELECT events_partition_key, aggregate_id
-                FROM aggregates
+              'SELECT aggregate_id
+                 FROM aggregates
                 WHERE ($1::text IS NULL OR aggregate_type_id = (SELECT id FROM aggregate_types WHERE type = $1))
-                AND ((events_partition_key >= $3 AND $4::uuid IS NULL)
-                     OR (events_partition_key, aggregate_id) > ($3, $4))
-                ORDER BY 1, 2
+                  AND ($3::uuid IS NULL OR aggregate_id > $3)
+                ORDER BY 1
                 LIMIT $2',
               'aggregates_to_update',
               [
                 aggregate_type,
                 group_size,
-                last_events_partition_key,
                 last_aggregate_id,
               ],
             ).to_a
             break if aggregate_rows.empty?
 
-            last_events_partition_key = aggregate_rows.last['events_partition_key']
             last_aggregate_id = aggregate_rows.last['aggregate_id']
 
             yielder << aggregate_rows.map { |x| x['aggregate_id'] }

spec/lib/sequent/core/event_store_spec.rb

@@ -407,8 +407,8 @@ class MyAggregate < Sequent::Core::AggregateRoot
       end
       let(:ordered_aggregate_ids) do
         event_streams
-          .sort_by { |s| [s.events_partition_key, s.aggregate_id] }
           .map(&:aggregate_id)
+          .sort
       end
 
       let(:group_size) { 100 }
@@ -435,7 +435,12 @@ class MyAggregate < Sequent::Core::AggregateRoot
         it 'finds all event streams of a specific type' do
           subject = event_store.event_streams_enumerator(aggregate_type: 'MyAggregate0', group_size:)
           aggregate_ids = subject.next
-          expect(aggregate_ids).to eq(ordered_aggregate_ids[0...10])
+          expect(aggregate_ids).to eq(
+            event_streams
+              .select { |s| s.aggregate_type == 'MyAggregate0' }
+              .map(&:aggregate_id)
+              .sort,
+          )
           expect { subject.next }.to raise_error(StopIteration)
         end
       end
@@ -455,7 +460,12 @@ class MyAggregate < Sequent::Core::AggregateRoot
         it 'finds all event streams of a specific type' do
           subject = event_store.event_streams_enumerator(aggregate_type: 'MyAggregate1', group_size:)
           aggregate_ids = subject.next
-          expect(aggregate_ids).to eq(ordered_aggregate_ids[10..])
+          expect(aggregate_ids).to eq(
+            event_streams
+              .select { |s| s.aggregate_type == 'MyAggregate1' }
+              .map(&:aggregate_id)
+              .sort,
+          )
           expect { subject.next }.to raise_error(StopIteration)
         end
       end