Fix: #54 Fix pagination in GetStreamsToSnapshot for large datasets

Address issue #54 in NEventStore.Persistence.Sql where
GetStreamsToSnapshot failed when the number of streams
exceeded PageSize. Updated SQL queries to include
`C.StreamId > @StreamId` for proper pagination.

Added `_lastStreamId` tracking in tests to validate the
pagination logic. Introduced new test cases to ensure
`GetFrom` and `GetFromAsync` methods handle large datasets
without crashing.

Updated SQL dialect resources and persistence engine
methods to support the new pagination logic.

Author: AGiorgetti

Changelog.md

@@ -7,6 +7,10 @@
   - Per-stream operations and delete: (BucketId, StreamId)
   - Snapshot join/aggregation: (StreamId, StreamRevision)
 
+### BugFix
+
+- Fixed GetStreamsToSnapshot pagination [#54](https://github.com/NEventStore/NEventStore.Persistence.SQL/issues/54)
+
 ### Breaking Changes
 
 The following indexes were added to the default schema (upon database initialization), add it manually to an existing database:

src/AdditionalTests/AdditionalPersistenceTests.Async.cs

@@ -20,20 +20,37 @@
 namespace NEventStore.Persistence.AcceptanceTests.Async
 {
 
+    /// <summary>
+    /// <para>
+    /// NEventStore.Persistence.Sql issue #54
+    /// GetStreamsToSnapshot fails when number of streams exceeds PageSize.
+    /// </para>
+    /// <para>
+    /// Query implementation was wrong: checking how the query and the pagination were implemented
+    /// it was lacking to check for C.StreamId > @StreamId.
+    /// which is needed to page through the results (similar to @Coalesce in GetCommitsFromStartingRevision)
+    /// </para>
+    /// </summary>
 #if MSTEST
     [TestClass]
 #endif
     public class when_getting_streams_to_snapshot_amount_exceeds_PageSize : PersistenceEngineConcernAsync
     {
         private int _moreThanPageSize;
+        private Guid _lastStreamId;
 
         protected override async Task BecauseAsync()
         {
             _moreThanPageSize = ConfiguredPageSizeForTesting + 1;
             var eventStore = new OptimisticEventStore(Persistence, null, null);
             for (int i = 0; i < _moreThanPageSize; i++)
             {
-                using IEventStream stream = await eventStore.OpenStreamAsync(Guid.NewGuid()).ConfigureAwait(false);
+                _lastStreamId = Guid.NewGuid();
+                using IEventStream stream = await eventStore.OpenStreamAsync(_lastStreamId).ConfigureAwait(false);
+                stream.Add(new EventMessage { Body = new TestEvent() { S = "Hi " + i } });
+                await stream.CommitChangesAsync(Guid.NewGuid(), CancellationToken.None).ConfigureAwait(false);
+                stream.Add(new EventMessage { Body = new TestEvent() { S = "Hi " + i } });
+                await stream.CommitChangesAsync(Guid.NewGuid(), CancellationToken.None).ConfigureAwait(false);
                 stream.Add(new EventMessage { Body = new TestEvent() { S = "Hi " + i } });
                 await stream.CommitChangesAsync(Guid.NewGuid(), CancellationToken.None).ConfigureAwait(false);
             }
@@ -49,6 +66,17 @@ public void GetStreamsToSnapshot_does_not_crash_and_returns_all_the_streams()
                 observer.StreamHeads.Count.Should().Be(_moreThanPageSize);
             });
         }
+
+        [Fact]
+        public void GetFrom_does_not_crash_and_returns_all_the_streams()
+        {
+            Assert.DoesNotThrowAsync(async () =>
+            {
+                var observer = new CommitStreamObserver();
+                await Persistence.GetFromAsync(Bucket.Default, _lastStreamId.ToString(), 0, int.MaxValue, observer, CancellationToken.None).ConfigureAwait(false);
+                observer.Commits.Count.Should().Be(3);
+            });
+        }
     }
 
 }

src/AdditionalTests/AdditionalPersistenceTests.cs

@@ -20,20 +20,37 @@
 namespace NEventStore.Persistence.AcceptanceTests
 {
 
+    /// <summary>
+    /// <para>
+    /// NEventStore.Persistence.Sql issue #54
+    /// GetStreamsToSnapshot fails when number of streams exceeds PageSize.
+    /// </para>
+    /// <para>
+    /// Query implementation was wrong: checking how the query and the pagination were implemented
+    /// it was lacking to check for C.StreamId > @StreamId.
+    /// which is needed to page through the results (similar to @Coalesce in GetCommitsFromStartingRevision)
+    /// </para>
+    /// </summary>
 #if MSTEST
     [TestClass]
 #endif
     public class when_getting_streams_to_snapshot_amount_exceeds_PageSize : PersistenceEngineConcern
     {
         private int _moreThanPageSize;
+        private Guid _lastStreamId;
 
         protected override void Because()
         {
             _moreThanPageSize = ConfiguredPageSizeForTesting + 1;
             var eventStore = new OptimisticEventStore(Persistence, null, null);
             for (int i = 0; i < _moreThanPageSize; i++)
             {
-                using IEventStream stream = eventStore.OpenStream(Guid.NewGuid());
+                _lastStreamId = Guid.NewGuid();
+                using IEventStream stream = eventStore.OpenStream(_lastStreamId);
+                stream.Add(new EventMessage { Body = new TestEvent() { S = "Hi " + i } });
+                stream.CommitChanges(Guid.NewGuid());
+                stream.Add(new EventMessage { Body = new TestEvent() { S = "Hi " + i } });
+                stream.CommitChanges(Guid.NewGuid());
                 stream.Add(new EventMessage { Body = new TestEvent() { S = "Hi " + i } });
                 stream.CommitChanges(Guid.NewGuid());
             }
@@ -48,6 +65,16 @@ public void GetStreamsToSnapshot_does_not_crash_and_returns_all_the_streams()
                 streamHeads.Length.Should().Be(_moreThanPageSize);
             });
         }
+
+        [Fact]
+        public void GetFrom_does_not_crash_and_returns_all_the_streams()
+        {
+            Assert.DoesNotThrow(() =>
+            {
+                var streamHeads = Persistence.GetFrom(Bucket.Default, _lastStreamId.ToString(), 0, int.MaxValue).ToArray();
+                streamHeads.Length.Should().Be(3);
+            });
+        }
     }
 
 }

src/NEventStore.Persistence.Sql/SqlDialects/CommonSqlStatements.Designer.cs

@@ -188,6 +188,7 @@ AND NOT EXISTS
    AND C.StreamRevision >= S.StreamRevision
  GROUP BY C.StreamId, C.BucketId, C.StreamIdOriginal
 HAVING MAX(C.StreamRevision) >= MAX(COALESCE(S.StreamRevision, 0)) + @Threshold
+   AND C.StreamId > @StreamId
  ORDER BY C.StreamId
  LIMIT @Limit;</value>
   </data>

src/NEventStore.Persistence.Sql/SqlDialects/CommonSqlStatements.resx

@@ -211,6 +211,7 @@ FROM (
     ON C.BucketId = :BucketId AND C.StreamId = S.StreamId AND C.StreamRevision >= S.StreamRevision
   GROUP BY C.StreamId, C.BucketId, C.StreamIdOriginal
   HAVING MAX(C.StreamRevision) >= MAX(COALESCE(S.StreamRevision, 0)) + :Threshold
+   AND C.StreamId > :StreamId
   ORDER BY C.StreamId
 )
 WHERE ROWNUM &lt;= :Limit;</value>

src/NEventStore.Persistence.Sql/SqlDialects/OracleNativeStatements.resx

@@ -259,6 +259,7 @@ public Task GetStreamsToSnapshotAsync(string bucketId, int maxThreshold, IAsyncO
 				string statement = _dialect.GetStreamsRequiringSnapshots;
 				query.AddParameter(_dialect.BucketId, bucketId, DbType.AnsiString);
 				query.AddParameter(_dialect.Threshold, maxThreshold);
+				query.AddParameter(_dialect.StreamId, String.Empty, DbType.AnsiString); // (NEventStore.Persistence.Sql #54) to define the parameter for paging
 				return query.ExecutePagedQueryAsync(
 					statement, (q, s) => q.SetParameter(_dialect.StreamId, _dialect.CoalesceParameterValue(s.StreamId()), DbType.AnsiString), // todo: I'm not sure this is used, the query does not have a "StreamId" parameter
 					asyncObserver, cancellationToken);

src/NEventStore.Persistence.Sql/SqlPersistenceEngine.Async.cs

@@ -210,9 +210,10 @@ public virtual IEnumerable<IStreamHead> GetStreamsToSnapshot(string bucketId, in
 					string statement = _dialect.GetStreamsRequiringSnapshots;
 					query.AddParameter(_dialect.BucketId, bucketId, DbType.AnsiString);
 					query.AddParameter(_dialect.Threshold, maxThreshold);
+					query.AddParameter(_dialect.StreamId, String.Empty, DbType.AnsiString); // (NEventStore.Persistence.Sql #54) to define the parameter for paging
 					return
 						query.ExecutePagedQuery(statement,
-							(q, s) => { }) // No next page delegate needed - @Skip parameter is handled automatically by PagedEnumerationCollection
+							(q, s) => q.SetParameter(_dialect.StreamId, _dialect.CoalesceParameterValue(s.StreamId()), DbType.AnsiString))
 							.Select(x => x.GetStreamToSnapshot());
 				});
 		}